YQ-4735 Json format settings / skip jsons errors / to stable (ydb-platform#28564)

kardymonds · GrigoriyPA · web-flow · commit e21ecb02a58f · 2025-11-12T11:54:58.000+03:00
Co-authored-by: Pisarenko Grigoriy &lt;grigoriypisar@ydb.tech&gt;
diff --git a/ydb/core/external_sources/object_storage.cpp b/ydb/core/external_sources/object_storage.cpp
@@ -262,6 +262,10 @@ struct TObjectStorageExternalSource : public IExternalSource {
                 continue;
             }
 
+            if (key == "skip.json.errors"sv) {
+                continue;
+            }
+
             if (matchAllSettings) {
                 issues.AddIssue(MakeErrorIssue(Ydb::StatusIds::BAD_REQUEST, "unknown format setting " + key));
             }
diff --git a/ydb/core/fq/libs/config/protos/row_dispatcher.proto b/ydb/core/fq/libs/config/protos/row_dispatcher.proto
@@ -23,7 +23,6 @@ message TJsonParserConfig {
     uint64 BatchSizeBytes = 1;  // default 1 MiB
     uint64 BatchCreationTimeoutMs = 2;
     uint64 BufferCellCount = 3;  // (number rows) * (number columns) limit, default 10^6
-    bool SkipErrors = 4;
 }
 
 message TCompileServiceConfig {
diff --git a/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.cpp b/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.cpp
@@ -678,10 +678,10 @@ ITopicFormatHandler::TPtr CreateTopicFormatHandler(const NActors::TActorContext&
     return ITopicFormatHandler::TPtr(handler);
 }
 
-TFormatHandlerConfig CreateFormatHandlerConfig(const NConfig::TRowDispatcherConfig& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId) {
+TFormatHandlerConfig CreateFormatHandlerConfig(const NConfig::TRowDispatcherConfig& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId, bool skipJsonErrors) {
     return {
         .FunctionRegistry = functionRegistry,
-        .JsonParserConfig = CreateJsonParserConfig(rowDispatcherConfig.GetJsonParser(), functionRegistry),
+        .JsonParserConfig = CreateJsonParserConfig(rowDispatcherConfig.GetJsonParser(), functionRegistry, skipJsonErrors),
         .FiltersConfig = {
             .CompileServiceId = compileServiceId
         }
diff --git a/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.h b/ydb/core/fq/libs/row_dispatcher/format_handler/format_handler.h
@@ -81,7 +81,7 @@ struct TFormatHandlerConfig {
 };
 
 ITopicFormatHandler::TPtr CreateTopicFormatHandler(const NActors::TActorContext& owner, const TFormatHandlerConfig& config, const ITopicFormatHandler::TSettings& settings, const TCountersDesc& counters);
-TFormatHandlerConfig CreateFormatHandlerConfig(const NConfig::TRowDispatcherConfig& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId);
+TFormatHandlerConfig CreateFormatHandlerConfig(const NConfig::TRowDispatcherConfig& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId, bool skipJsonErrors);
 
 namespace NTests {
 
diff --git a/ydb/core/fq/libs/row_dispatcher/format_handler/parsers/json_parser.cpp b/ydb/core/fq/libs/row_dispatcher/format_handler/parsers/json_parser.cpp
@@ -736,7 +736,7 @@ TValueStatus<ITopicParser::TPtr> CreateJsonParser(IParsedDataConsumer::TPtr cons
     return ITopicParser::TPtr(parser);
 }
 
-TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry) {
+TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, bool skipErrors) {
     TJsonParserConfig result = {.FunctionRegistry = functionRegistry};
     if (const auto batchSize = parserConfig.GetBatchSizeBytes()) {
         result.BatchSize = batchSize;
@@ -745,7 +745,7 @@ TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parse
         result.BufferCellCount = bufferCellCount;
     }
     result.LatencyLimit = TDuration::MilliSeconds(parserConfig.GetBatchCreationTimeoutMs());
-    result.SkipErrors = parserConfig.GetSkipErrors();
+    result.SkipErrors = skipErrors;
     return result;
 }
 
diff --git a/ydb/core/fq/libs/row_dispatcher/format_handler/parsers/json_parser.h b/ydb/core/fq/libs/row_dispatcher/format_handler/parsers/json_parser.h
@@ -23,6 +23,6 @@ struct TJsonParserConfig {
 };
 
 TValueStatus<ITopicParser::TPtr> CreateJsonParser(IParsedDataConsumer::TPtr consumer, const TJsonParserConfig& config, const TCountersDesc& counters);
-TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry);
+TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, bool skipErrors);
 
 }  // namespace NFq::NRowDispatcher
diff --git a/ydb/core/fq/libs/row_dispatcher/topic_session.cpp b/ydb/core/fq/libs/row_dispatcher/topic_session.cpp
@@ -263,7 +263,8 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
     const NYql::IPqGateway::TPtr PqGateway;
     const std::shared_ptr<NYdb::ICredentialsProviderFactory> CredentialsProviderFactory;
     const NConfig::TRowDispatcherConfig Config;
-    const TFormatHandlerConfig FormatHandlerConfig;
+    const TActorId CompileServiceActorId;
+    const NKikimr::NMiniKQL::IFunctionRegistry* FunctionRegistry;
     const i64 BufferSize;
     TString LogPrefix;
 
@@ -281,6 +282,7 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
     ui64 QueuedBytes = 0;
     TMaybe<TString> ConsumerName;
     TInstant StartingMessageTimestamp;
+    TMaybe<bool> SkipJsonErrors;
 
     // Metrics
     TInstant WaitEventStartedAt;
@@ -401,7 +403,8 @@ TTopicSession::TTopicSession(
     , PqGateway(pqGateway)
     , CredentialsProviderFactory(credentialsProviderFactory)
     , Config(config)
-    , FormatHandlerConfig(CreateFormatHandlerConfig(config, functionRegistry, compileServiceActorId))
+    , CompileServiceActorId(compileServiceActorId)
+    , FunctionRegistry(functionRegistry)
     , BufferSize(maxBufferSize)
     , LogPrefix("TopicSession")
     , Counters(counters)
@@ -767,9 +770,10 @@ void TTopicSession::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
     auto clientInfo = Clients.insert({ev->Sender, MakeIntrusive<TClientsInfo>(*this, LogPrefix, handlerSettings, ev, Counters, ReadGroup, offset)}).first->second;
     auto formatIt = FormatHandlers.find(handlerSettings);
     if (formatIt == FormatHandlers.end()) {
+        auto config = CreateFormatHandlerConfig(Config, FunctionRegistry, CompileServiceActorId, source.GetSkipJsonErrors());
         formatIt = FormatHandlers.emplace(handlerSettings, CreateTopicFormatHandler(
             ActorContext(),
-            FormatHandlerConfig,
+            config,
             handlerSettings,
             {.CountersRoot = CountersRoot, .CountersSubgroup = Metrics.PartitionGroup}
         )).first;
@@ -781,6 +785,7 @@ void TTopicSession::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
     }
 
     ConsumerName = source.GetConsumerName();
+    SkipJsonErrors = source.GetSkipJsonErrors();
     SendStatistics();
 }
 
@@ -965,6 +970,11 @@ bool TTopicSession::CheckNewClient(NFq::TEvRowDispatcher::TEvStartSession::TPtr&
         return false;
     }
 
+    if (SkipJsonErrors && SkipJsonErrors != source.GetSkipJsonErrors()) {
+        LOG_ROW_DISPATCHER_INFO("Different skip json errors mode, expected " <<  SkipJsonErrors << ", actual " << source.GetSkipJsonErrors() << ", send error");
+        SendSessionError(ev->Sender, TStatus::Fail(EStatusId::PRECONDITION_FAILED, TStringBuilder() << "Use the same skip json errors settings in all queries via RD (current mode " << SkipJsonErrors << ")"), false);
+        return false;
+    }
     return true;
 }
 
diff --git a/ydb/core/fq/libs/row_dispatcher/ut/topic_session_ut.cpp b/ydb/core/fq/libs/row_dispatcher/ut/topic_session_ut.cpp
@@ -48,13 +48,12 @@ class TFixture : public NTests::TBaseFixture {
         RowDispatcherActorId = Runtime.AllocateEdgeActor();
     }
 
-    void Init(const TString& topicPath, ui64 maxSessionUsedMemory = std::numeric_limits<ui64>::max(), bool skipErrors = false) {
+    void Init(const TString& topicPath, ui64 maxSessionUsedMemory = std::numeric_limits<ui64>::max()) {
         TopicPath = topicPath;
         Config.SetTimeoutBeforeStartSessionSec(TimeoutBeforeStartSessionSec);
         Config.SetMaxSessionUsedMemory(maxSessionUsedMemory);
         Config.SetSendStatusPeriodSec(2);
         Config.SetWithoutConsumer(false);
-        Config.MutableJsonParser()->SetSkipErrors(skipErrors);
         Config.MutableJsonParser()->SetBatchCreationTimeoutMs(100);
 
         auto credFactory = NKikimr::CreateYdbCredentialsProviderFactory;
@@ -122,7 +121,7 @@ class TFixture : public NTests::TBaseFixture {
         }
     }
 
-    NYql::NPq::NProto::TDqPqTopicSource BuildSource(bool emptyPredicate = false, const TString& consumer = DefaultPqConsumer) {
+    NYql::NPq::NProto::TDqPqTopicSource BuildSource(bool emptyPredicate = false, const TString& consumer = DefaultPqConsumer, bool skipErrors = false) {
         NYql::NPq::NProto::TDqPqTopicSource settings;
         settings.SetEndpoint(GetDefaultPqEndpoint());
         settings.SetTopicPath(TopicPath);
@@ -137,6 +136,7 @@ class TFixture : public NTests::TBaseFixture {
         if (!emptyPredicate) {
             settings.SetPredicate("TRUE");
         }
+        settings.SetSkipJsonErrors(skipErrors);
         return settings;
     }
 
@@ -648,8 +648,8 @@ Y_UNIT_TEST_SUITE(TopicSessionTests) {
     Y_UNIT_TEST_F(WrongJson, TRealTopicFixture) {
         const TString topicName = "wrong_json";
         PQCreateStream(topicName);
-        Init(topicName, std::numeric_limits<ui64>::max(), true);
-        auto source = BuildSource();
+        Init(topicName);
+        auto source = BuildSource(false, DefaultPqConsumer, true);
         StartSession(ReadActorId1, source);
         
         auto writeRead = [&](const std::vector<TString>& input, const TBatch& output) {
@@ -686,8 +686,8 @@ Y_UNIT_TEST_SUITE(TopicSessionTests) {
     Y_UNIT_TEST_F(WrongJsonOffset, TRealTopicFixture) {
         const TString topicName = "wrong_json_offset";
         PQCreateStream(topicName);
-        Init(topicName, std::numeric_limits<ui64>::max(), true);
-        auto source = BuildSource();
+        Init(topicName);
+        auto source = BuildSource(false, DefaultPqConsumer, true);
         StartSession(ReadActorId1, source);
 
         TString wrongJson{"wrong"};
diff --git a/ydb/library/yql/providers/pq/async_io/dq_pq_rd_read_actor.cpp b/ydb/library/yql/providers/pq/async_io/dq_pq_rd_read_actor.cpp
@@ -539,9 +539,8 @@ TDqPqRdReadActor::TDqPqRdReadActor(
         , CredentialsProviderFactory(std::move(credentialsProviderFactory))
         , MaxBufferSize(bufferSize)
 {
-
     SRC_LOG_I("Start read actor, local row dispatcher " << LocalRowDispatcherActorId.ToString() << ", metadatafields: " << JoinSeq(',', SourceParams.GetMetadataFields())
-        << ", partitions: " << JoinSeq(',', GetPartitionsToRead()));
+        << ", partitions: " << JoinSeq(',', GetPartitionsToRead()) << ", skip json errors: " << SourceParams.GetSkipJsonErrors());
     if (Parent != this) {
         return;
     }
diff --git a/ydb/library/yql/providers/pq/common/yql_names.h b/ydb/library/yql/providers/pq/common/yql_names.h
@@ -18,5 +18,6 @@ constexpr TStringBuf WatermarksLateArrivalDelayUsSetting = "WatermarksLateArriva
 constexpr TStringBuf WatermarksIdlePartitionsSetting = "WatermarksIdlePartitions";
 constexpr TStringBuf ReconnectPeriod = "ReconnectPeriod";
 constexpr TStringBuf ReadGroup = "ReadGroup";
+constexpr TStringBuf SkipJsonErrors = "SkipJsonErrors";
 
 } // namespace NYql
diff --git a/ydb/library/yql/providers/pq/proto/dq_io.proto b/ydb/library/yql/providers/pq/proto/dq_io.proto
@@ -90,6 +90,8 @@ message TDqPqTopicSource {
     repeated TTaskSensorLabel TaskSensorLabel = 23;
     string WatermarkExpr = 24;
     repeated uint64 NodeIds = 25;
+    bool UseActorSystemThreadsInTopicClient = 26;
+    bool SkipJsonErrors = 27;
 }
 
 message TDqPqTopicSink {
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_datasource.cpp b/ydb/library/yql/providers/pq/provider/yql_pq_datasource.cpp
@@ -184,6 +184,10 @@ class TPqDataSourceProvider : public TDataProviderBase {
             settings.Add(topicKeyParser.GetDateFormat());
         }
 
+        if (topicKeyParser.GetSkipJsonErrors()) {
+            settings.Add(topicKeyParser.GetSkipJsonErrors());
+        }
+
         auto builder = Build<TPqReadTopic>(ctx, read.Pos())
             .World(read.World())
             .DataSource(read.DataSource())
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp b/ydb/library/yql/providers/pq/provider/yql_pq_datasource_type_ann.cpp
@@ -286,6 +286,19 @@ class TPqDataSourceTypeAnnotationTransformer : public TVisitorTransformerBase {
             }
         }
 
+        if (const auto maybeSkipJsonErrorsSetting = FindSetting(settings, SkipJsonErrors)) {
+            const auto value = maybeSkipJsonErrorsSetting.Cast().Ptr();
+            if (!EnsureAtom(*value, ctx)) {
+                return TStatus::Error;
+            }
+            bool skipJsonErrorsSetting;
+            if (!TryFromString<bool>(value->Content(), skipJsonErrorsSetting)) {
+                ctx.AddError(TIssue(ctx.GetPosition(settings->Pos()), TStringBuilder()
+                    << "Expected bool, but got: " << value->Content()));
+                return TStatus::Error;
+            }
+        }
+
         if (const auto maybeSharedReadingSetting = FindSetting(settings, SharedReading)) {
             const auto value = maybeSharedReadingSetting.Cast().Ptr();
             if (!EnsureAtom(*value, ctx)) {
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_dq_integration.cpp b/ydb/library/yql/providers/pq/provider/yql_pq_dq_integration.cpp
@@ -116,13 +116,32 @@ class TPqDqIntegration: public TDqIntegrationBase {
                 }
             }
 
+            bool skipJsonErrors = false;
+            if (auto settingsList = pqReadTopic.Settings().Maybe<TExprList>()) {
+                for (const TExprNode::TPtr& s : pqReadTopic.Settings().Raw()->Children()) {
+                    if (s->ChildrenSize() >= 2 && s->Child(0)->Content() == "skip.json.errors"sv) {
+                        if (!TryFromString<bool>(s->Child(1)->Content(), skipJsonErrors)) {
+                            ctx.AddError(TIssue(ctx.GetPosition(pqReadTopic.Pos()), R"("skip.json.errors" must be boolean type))"));
+                            return {};
+                        }
+                    }
+                }
+            }
+
+            if (skipJsonErrors) {
+                auto clusterConfiguration = GetClusterConfiguration(clusterName);
+                if (!UseSharedReading(clusterConfiguration, format) ) {
+                    ctx.AddError(TIssue(ctx.GetPosition(pqReadTopic.Pos()), R"("skip.json.errors" is supported only in shared reading mode))"));
+                    return {};
+                }
+            }
 
             return Build<TDqSourceWrap>(ctx, pos)
                 .Input<TDqPqTopicSource>()
                     .World(pqReadTopic.World())
                     .Topic(pqReadTopic.Topic())
                     .Columns(std::move(columnNames))
-                    .Settings(BuildTopicReadSettings(clusterName, wrSettings, pos, format, ctx))
+                    .Settings(BuildTopicReadSettings(clusterName, wrSettings, pos, format, skipJsonErrors, ctx))
                     .Token<TCoSecureParam>()
                         .Name().Build(token)
                         .Build()
@@ -197,6 +216,7 @@ class TPqDqIntegration: public TDqIntegrationBase {
                 }
 
                 bool sharedReading = false;
+                bool skipErrors = false;
                 TString format;
                 size_t const settingsCount = topicSource.Settings().Size();
                 for (size_t i = 0; i < settingsCount; ++i) {
@@ -226,6 +246,8 @@ class TPqDqIntegration: public TDqIntegrationBase {
                         srcDesc.MutableWatermarks()->SetLateArrivalDelayUs(FromString<ui64>(Value(setting)));
                     } else if (name == WatermarksIdlePartitionsSetting) {
                         srcDesc.MutableWatermarks()->SetIdlePartitionsEnabled(true);
+                    } else if (name == SkipJsonErrors) {
+                        skipErrors = FromString<bool>(Value(setting));
                     }
                 }
 
@@ -274,6 +296,8 @@ class TPqDqIntegration: public TDqIntegrationBase {
                     srcDesc.SetPredicate(predicateSql);
                     srcDesc.SetSharedReading(true);
                 }
+                srcDesc.SetSkipJsonErrors(skipErrors);
+
                 *srcDesc.MutableDisposition() = State_->Disposition;
 
                 for (const auto& [label, value] : State_->TaskSensorLabels) {
@@ -358,6 +382,7 @@ class TPqDqIntegration: public TDqIntegrationBase {
         const IDqIntegration::TWrapReadSettings& wrSettings,
         TPositionHandle pos,
         std::string_view format,
+        bool skipJsonErrors,
         TExprContext& ctx) const
     {
         TVector<TCoNameValueTuple> props;
@@ -376,6 +401,9 @@ class TPqDqIntegration: public TDqIntegrationBase {
         Add(props, ReconnectPeriod, ToString(clusterConfiguration->ReconnectPeriod), pos, ctx);
         Add(props, Format, format, pos, ctx);
         Add(props, ReadGroup, clusterConfiguration->ReadGroup, pos, ctx);
+        if (skipJsonErrors) {
+            Add(props, SkipJsonErrors, ToString(skipJsonErrors), pos, ctx);
+        }
 
         if (clusterConfiguration->UseSsl) {
             Add(props, UseSslSetting, "1", pos, ctx);
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_topic_key_parser.cpp b/ydb/library/yql/providers/pq/provider/yql_pq_topic_key_parser.cpp
@@ -75,6 +75,10 @@ bool TTopicKeyParser::Parse(const TExprNode& expr, TExprNode::TPtr readSettings,
                 Watermark = readSettings->Child(i)->ChildPtr(1);
                 continue;
             }
+            if (readSettings->Child(i)->Head().IsAtom("skip.json.errors")) {
+                SkipJsonErrors = readSettings->Child(i);
+                continue;
+            }
         }
     }
 
diff --git a/ydb/library/yql/providers/pq/provider/yql_pq_topic_key_parser.h b/ydb/library/yql/providers/pq/provider/yql_pq_topic_key_parser.h
@@ -54,6 +54,10 @@ class TTopicKeyParser {
         return Watermark;
     }
 
+    TExprNode::TPtr GetSkipJsonErrors() const {
+        return SkipJsonErrors;
+    }
+
     bool Parse(const TExprNode& expr, TExprNode::TPtr readSettings, TExprContext& ctx);
 
 private:
@@ -72,6 +76,7 @@ class TTopicKeyParser {
     TExprNode::TPtr UserSchema;
     TExprNode::TPtr ColumnOrder;
     TExprNode::TPtr Watermark;
+    TExprNode::TPtr SkipJsonErrors;
 };
 
 } // namespace NYql
diff --git a/ydb/tests/fq/yds/test_row_dispatcher.py b/ydb/tests/fq/yds/test_row_dispatcher.py

Original file line number	Diff line number	Diff line change
`@@ -262,6 +262,10 @@ struct TObjectStorageExternalSource : public IExternalSource {`
`262`	`262`	`continue;`
`263`	`263`	`}`
`264`	`264`
	`265`	`+ if (key == "skip.json.errors"sv) {`
	`266`	`+ continue;`
	`267`	`+ }`
	`268`	`+`
`265`	`269`	`if (matchAllSettings) {`
`266`	`270`	`issues.AddIssue(MakeErrorIssue(Ydb::StatusIds::BAD_REQUEST, "unknown format setting " + key));`
`267`	`271`	`}`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,6 @@ message TJsonParserConfig {`
`23`	`23`	`uint64 BatchSizeBytes = 1; // default 1 MiB`
`24`	`24`	`uint64 BatchCreationTimeoutMs = 2;`
`25`	`25`	`uint64 BufferCellCount = 3; // (number rows) * (number columns) limit, default 10^6`
`26`		`- bool SkipErrors = 4;`
`27`	`26`	`}`
`28`	`27`
`29`	`28`	`message TCompileServiceConfig {`
Original file line number	Diff line number	Diff line change
`@@ -678,10 +678,10 @@ ITopicFormatHandler::TPtr CreateTopicFormatHandler(const NActors::TActorContext&`
`678`	`678`	`return ITopicFormatHandler::TPtr(handler);`
`679`	`679`	`}`
`680`	`680`
`681`		`-TFormatHandlerConfig CreateFormatHandlerConfig(const NConfig::TRowDispatcherConfig& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId) {`
	`681`	`+TFormatHandlerConfig CreateFormatHandlerConfig(const NConfig::TRowDispatcherConfig& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId, bool skipJsonErrors) {`
`682`	`682`	`return {`
`683`	`683`	`.FunctionRegistry = functionRegistry,`
`684`		`- .JsonParserConfig = CreateJsonParserConfig(rowDispatcherConfig.GetJsonParser(), functionRegistry),`
	`684`	`+ .JsonParserConfig = CreateJsonParserConfig(rowDispatcherConfig.GetJsonParser(), functionRegistry, skipJsonErrors),`
`685`	`685`	`.FiltersConfig = {`
`686`	`686`	`.CompileServiceId = compileServiceId`
`687`	`687`	`}`
Original file line number	Diff line number	Diff line change
`@@ -736,7 +736,7 @@ TValueStatus<ITopicParser::TPtr> CreateJsonParser(IParsedDataConsumer::TPtr cons`
`736`	`736`	`return ITopicParser::TPtr(parser);`
`737`	`737`	`}`
`738`	`738`
`739`		`-TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry) {`
	`739`	`+TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, bool skipErrors) {`
`740`	`740`	`TJsonParserConfig result = {.FunctionRegistry = functionRegistry};`
`741`	`741`	`if (const auto batchSize = parserConfig.GetBatchSizeBytes()) {`
`742`	`742`	`result.BatchSize = batchSize;`
`@@ -745,7 +745,7 @@ TJsonParserConfig CreateJsonParserConfig(const NConfig::TJsonParserConfig& parse`
`745`	`745`	`result.BufferCellCount = bufferCellCount;`
`746`	`746`	`}`
`747`	`747`	`result.LatencyLimit = TDuration::MilliSeconds(parserConfig.GetBatchCreationTimeoutMs());`
`748`		`- result.SkipErrors = parserConfig.GetSkipErrors();`
	`748`	`+ result.SkipErrors = skipErrors;`
`749`	`749`	`return result;`
`750`	`750`	`}`
`751`	`751`
Original file line number	Diff line number	Diff line change
`@@ -539,9 +539,8 @@ TDqPqRdReadActor::TDqPqRdReadActor(`
`539`	`539`	`, CredentialsProviderFactory(std::move(credentialsProviderFactory))`
`540`	`540`	`, MaxBufferSize(bufferSize)`
`541`	`541`	`{`
`542`		`-`
`543`	`542`	`SRC_LOG_I("Start read actor, local row dispatcher " << LocalRowDispatcherActorId.ToString() << ", metadatafields: " << JoinSeq(',', SourceParams.GetMetadataFields())`
`544`		`- << ", partitions: " << JoinSeq(',', GetPartitionsToRead()));`
	`543`	`+ << ", partitions: " << JoinSeq(',', GetPartitionsToRead()) << ", skip json errors: " << SourceParams.GetSkipJsonErrors());`
`545`	`544`	`if (Parent != this) {`
`546`	`545`	`return;`
`547`	`546`	`}`