YQ-4735 Json format settings / skip jsons errors (ydb-platform#28329)

kardymonds · GrigoriyPA · qyryq · commit 4d2513565c31 · 2025-11-24T13:16:40.000Z
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
@@ -272,6 +272,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/common/row_dispatcher_settings.cpp b/ydb/core/fq/libs/row_dispatcher/common/row_dispatcher_settings.cpp
@@ -9,7 +9,6 @@ namespace NFq {
 
 TRowDispatcherSettings::TJsonParserSettings::TJsonParserSettings(const NConfig::TJsonParserConfig& config)
     : BatchCreationTimeout(TDuration::MilliSeconds(config.GetBatchCreationTimeoutMs()))
-    , SkipErrors(config.GetSkipErrors())
 {
     if (config.GetBatchSizeBytes()) {
         BatchSizeBytes = config.GetBatchSizeBytes();
diff --git a/ydb/core/fq/libs/row_dispatcher/common/row_dispatcher_settings.h b/ydb/core/fq/libs/row_dispatcher/common/row_dispatcher_settings.h
@@ -33,7 +33,6 @@ class TRowDispatcherSettings {
         YDB_ACCESSOR(ui64, BatchSizeBytes, 1_MB);
         YDB_ACCESSOR(TDuration, BatchCreationTimeout, TDuration::Seconds(1));
         YDB_ACCESSOR(ui64, BufferCellCount, 1000'000);
-        YDB_ACCESSOR(bool, SkipErrors, false);
     };
 
     class TCompileServiceSettings {
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
@@ -677,10 +677,10 @@ ITopicFormatHandler::TPtr CreateTopicFormatHandler(const NActors::TActorContext&
     return ITopicFormatHandler::TPtr(handler);
 }
 
-TFormatHandlerConfig CreateFormatHandlerConfig(const TRowDispatcherSettings& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId) {
+TFormatHandlerConfig CreateFormatHandlerConfig(const TRowDispatcherSettings& 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
@@ -78,7 +78,7 @@ struct TFormatHandlerConfig {
 };
 
 ITopicFormatHandler::TPtr CreateTopicFormatHandler(const NActors::TActorContext& owner, const TFormatHandlerConfig& config, const ITopicFormatHandler::TSettings& settings, const TCountersDesc& counters);
-TFormatHandlerConfig CreateFormatHandlerConfig(const TRowDispatcherSettings& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId);
+TFormatHandlerConfig CreateFormatHandlerConfig(const TRowDispatcherSettings& 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
@@ -743,13 +743,13 @@ TValueStatus<ITopicParser::TPtr> CreateJsonParser(IParsedDataConsumer::TPtr cons
     return ITopicParser::TPtr(parser);
 }
 
-TJsonParserConfig CreateJsonParserConfig(const TRowDispatcherSettings::TJsonParserSettings& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry) {
+TJsonParserConfig CreateJsonParserConfig(const TRowDispatcherSettings::TJsonParserSettings& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, bool skipErrors) {
     return {
         .FunctionRegistry = functionRegistry,
         .BatchSize = parserConfig.GetBatchSizeBytes(),
         .LatencyLimit = parserConfig.GetBatchCreationTimeout(),
         .BufferCellCount = parserConfig.GetBufferCellCount(),
-        .SkipErrors = parserConfig.GetSkipErrors()
+        .SkipErrors = skipErrors
     };
 }
 
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
@@ -21,6 +21,6 @@ struct TJsonParserConfig {
 };
 
 TValueStatus<ITopicParser::TPtr> CreateJsonParser(IParsedDataConsumer::TPtr consumer, const TJsonParserConfig& config, const TCountersDesc& counters);
-TJsonParserConfig CreateJsonParserConfig(const TRowDispatcherSettings::TJsonParserSettings& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry);
+TJsonParserConfig CreateJsonParserConfig(const TRowDispatcherSettings::TJsonParserSettings& 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
@@ -267,7 +267,8 @@ class TTopicSession : public TActorBootstrapped<TTopicSession>, NYql::TTopicEven
     const NYql::IPqGateway::TPtr PqGateway;
     const std::shared_ptr<NYdb::ICredentialsProviderFactory> CredentialsProviderFactory;
     const TRowDispatcherSettings Config;
-    const TFormatHandlerConfig FormatHandlerConfig;
+    const TActorId CompileServiceActorId;
+    const NKikimr::NMiniKQL::IFunctionRegistry* FunctionRegistry;
     const i64 BufferSize;
     TString LogPrefix;
 
@@ -285,6 +286,7 @@ class TTopicSession : public TActorBootstrapped<TTopicSession>, NYql::TTopicEven
     ui64 QueuedBytes = 0;
     TMaybe<TString> ConsumerName;
     TInstant StartingMessageTimestamp;
+    TMaybe<bool> SkipJsonErrors;
 
     // Metrics
     TInstant WaitEventStartedAt;
@@ -407,7 +409,8 @@ TTopicSession::TTopicSession(
     , PqGateway(pqGateway)
     , CredentialsProviderFactory(credentialsProviderFactory)
     , Config(config)
-    , FormatHandlerConfig(CreateFormatHandlerConfig(config, functionRegistry, compileServiceActorId))
+    , CompileServiceActorId(compileServiceActorId)
+    , FunctionRegistry(functionRegistry)
     , BufferSize(maxBufferSize)
     , LogPrefix("TopicSession")
     , Counters(counters)
@@ -783,9 +786,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;
@@ -797,6 +801,7 @@ void TTopicSession::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
     }
 
     ConsumerName = source.GetConsumerName();
+    SkipJsonErrors = source.GetSkipJsonErrors();
     SendStatistics();
 }
 
@@ -981,6 +986,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
@@ -49,13 +49,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;
@@ -124,7 +123,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);
@@ -139,6 +138,7 @@ class TFixture : public NTests::TBaseFixture {
         if (!emptyPredicate) {
             settings.SetPredicate("TRUE");
         }
+        settings.SetSkipJsonErrors(skipErrors);
         return settings;
     }
 
@@ -631,8 +631,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) {
@@ -669,8 +669,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
@@ -540,9 +540,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
@@ -91,6 +91,7 @@ message TDqPqTopicSource {
     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
@@ -298,6 +298,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()
@@ -218,6 +237,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) {
@@ -247,6 +267,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));
                     }
                 }
 
@@ -296,6 +318,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) {
@@ -390,6 +414,7 @@ class TPqDqIntegration: public TDqIntegrationBase {
         const IDqIntegration::TWrapReadSettings& wrSettings,
         TPositionHandle pos,
         std::string_view format,
+        bool skipJsonErrors,
         TExprContext& ctx) const
     {
         TVector<TCoNameValueTuple> props;
@@ -408,6 +433,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/streaming/test_streaming.py b/ydb/tests/fq/streaming/test_streaming.py
@@ -295,3 +295,34 @@ def test_read_topic_restore_state(self, kikimr):
         self.write_stream(data)
         expected_data = ['{"a_time":null,"b_time":1696849942500001,"c_time":1696849943000001}']
         assert self.read_stream(len(expected_data), topic_path=self.output_topic) == expected_data
+
+    def test_json_errors(self, kikimr):
+        sourceName = "test_json_errors" + ''.join(random.choices(string.ascii_letters + string.digits, k=8))
+        self.init_topics(sourceName, partitions_count=10)
+        self.create_source(kikimr, sourceName, True)
+
+        name = "query1"
+        sql = R'''
+            CREATE STREAMING QUERY `{query_name}` AS
+            DO BEGIN
+                $in = SELECT data FROM {source_name}.`{input_topic}`
+                WITH (
+                    FORMAT="json_each_row",
+                    `skip.json.errors` = "true",
+                    SCHEMA=(time UINT32 NOT NULL, data String NOT NULL));
+                INSERT INTO {source_name}.`{output_topic}` SELECT data FROM $in;
+            END DO;'''
+
+        query_id = "query_id"  # TODO
+        kikimr.YdbClient.query(sql.format(query_name=name, source_name=sourceName, input_topic=self.input_topic, output_topic=self.output_topic))
+        self.wait_completed_checkpoints(kikimr, query_id)
+
+        data = [
+            '{"time": 101, "data": "hello1"}',
+            '{"time": 102, "data": 7777}',
+            '{"time": 103, "data": "hello2"}'
+        ]
+        self.write_stream(data, partition_key="key")
+
+        expected = ['hello1', 'hello2']
+        assert self.read_stream(len(expected), topic_path=self.output_topic) == expected
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
`@@ -272,6 +272,10 @@ struct TObjectStorageExternalSource : public IExternalSource {`
`272`	`272`	`continue;`
`273`	`273`	`}`
`274`	`274`
	`275`	`+ if (key == "skip.json.errors"sv) {`
	`276`	`+ continue;`
	`277`	`+ }`
	`278`	`+`
`275`	`279`	`if (matchAllSettings) {`
`276`	`280`	`issues.AddIssue(MakeErrorIssue(Ydb::StatusIds::BAD_REQUEST, "unknown format setting " + key));`
`277`	`281`	`}`
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
`@@ -9,7 +9,6 @@ namespace NFq {`
`9`	`9`
`10`	`10`	`TRowDispatcherSettings::TJsonParserSettings::TJsonParserSettings(const NConfig::TJsonParserConfig& config)`
`11`	`11`	`: BatchCreationTimeout(TDuration::MilliSeconds(config.GetBatchCreationTimeoutMs()))`
`12`		`- , SkipErrors(config.GetSkipErrors())`
`13`	`12`	`{`
`14`	`13`	`if (config.GetBatchSizeBytes()) {`
`15`	`14`	`BatchSizeBytes = config.GetBatchSizeBytes();`
Original file line number	Diff line number	Diff line change
`@@ -677,10 +677,10 @@ ITopicFormatHandler::TPtr CreateTopicFormatHandler(const NActors::TActorContext&`
`677`	`677`	`return ITopicFormatHandler::TPtr(handler);`
`678`	`678`	`}`
`679`	`679`
`680`		`-TFormatHandlerConfig CreateFormatHandlerConfig(const TRowDispatcherSettings& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId) {`
	`680`	`+TFormatHandlerConfig CreateFormatHandlerConfig(const TRowDispatcherSettings& rowDispatcherConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, NActors::TActorId compileServiceId, bool skipJsonErrors) {`
`681`	`681`	`return {`
`682`	`682`	`.FunctionRegistry = functionRegistry,`
`683`		`- .JsonParserConfig = CreateJsonParserConfig(rowDispatcherConfig.GetJsonParser(), functionRegistry),`
	`683`	`+ .JsonParserConfig = CreateJsonParserConfig(rowDispatcherConfig.GetJsonParser(), functionRegistry, skipJsonErrors),`
`684`	`684`	`.FiltersConfig = {`
`685`	`685`	`.CompileServiceId = compileServiceId`
`686`	`686`	`}`
Original file line number	Diff line number	Diff line change
`@@ -743,13 +743,13 @@ TValueStatus<ITopicParser::TPtr> CreateJsonParser(IParsedDataConsumer::TPtr cons`
`743`	`743`	`return ITopicParser::TPtr(parser);`
`744`	`744`	`}`
`745`	`745`
`746`		`-TJsonParserConfig CreateJsonParserConfig(const TRowDispatcherSettings::TJsonParserSettings& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry) {`
	`746`	`+TJsonParserConfig CreateJsonParserConfig(const TRowDispatcherSettings::TJsonParserSettings& parserConfig, const NKikimr::NMiniKQL::IFunctionRegistry* functionRegistry, bool skipErrors) {`
`747`	`747`	`return {`
`748`	`748`	`.FunctionRegistry = functionRegistry,`
`749`	`749`	`.BatchSize = parserConfig.GetBatchSizeBytes(),`
`750`	`750`	`.LatencyLimit = parserConfig.GetBatchCreationTimeout(),`
`751`	`751`	`.BufferCellCount = parserConfig.GetBufferCellCount(),`
`752`		`- .SkipErrors = parserConfig.GetSkipErrors()`
	`752`	`+ .SkipErrors = skipErrors`
`753`	`753`	`};`
`754`	`754`	`}`
`755`	`755`