From 06d1850f3aa5e05a089c27624065c88e308ff340 Mon Sep 17 00:00:00 2001 From: Gabriel Date: Tue, 27 Oct 2026 10:07:35 +0800 Subject: [PATCH] [fix](external) Preserve ES residual filters and repair scanner regression setup ### What problem does this PR solve? Related PR: #69614 `external_table_p2/es/test_es_file_scanner_v2.groovy` fails on its first legacy-scanner baseline query with `Table [test1] does not exist in database [default_db]`. Its interpolated port key is a Groovy `GString`, while `otherConfigs` is a `Properties` containing `String` keys. `Properties.get(Object)` returns null even when the port is configured, producing `http://:null`. Convert the key to `String` and fail explicitly when the configuration is absent. The failure happens before any V2 result comparison or BE scan. The coverage review also found a query correctness issue shared by both scanners: a nested AND can return a partial ES DSL and record an unsupported child, but the single-expression path ignored that residual tracking. For `(a = 1 AND local_predicate) OR a = 2`, the engine could drop the local predicate; `NOT(a = 1 AND local_predicate)` could also become an unsafe `NOT(a = 1)` at the source. Retain the complete expression locally whenever conversion records a residual. Fully pushable expressions and independent top-level AND conjuncts retain their existing pushdown behavior. Expand coverage in the same change: - ES 5/6/7/8 V1/V2 comparisons across source/docvalue scans, parallel/whole-index scrolling, and batch sizes 1/1024. Add scalar/date/array/JSON, IN/NOT IN, null, raw ES query, residual filtering, aggregation and LIMIT cases; restore every changed session variable. - An independent equivalent-query oracle for nested residual OR, since a V1/V2 comparison alone cannot detect a shared FE error. Compare cardinality for unordered LIMIT rather than document identity across BEs. - FE tests for residual tracking, docvalue eligibility and field-count boundaries, alias/context serialization, shard planning and safe early-stop parameters. - BE tests for source-only fallback, missing/null docvalues and empty arrays, real HTTP empty results, early close and scroll 404/500 cleanup. Add an opt-in authenticated live-reader smoke test. ### Release note Preserve local filtering for Elasticsearch compound predicates that cannot be completely pushed down. ### Check List (For Author) - Test - [x] FE: all 127 ES connector tests pass. The two new nested OR/NOT tests fail before the residual-tracking fix. - [x] BE: all 27 selected ES ASAN tests pass against authenticated Elasticsearch 9.0.0 and 9.5.5, including source/docvalue reads with batch size 1. - [x] FE Checkstyle for the ES module and its reactor dependencies; clang-format 16, clang-tidy on changed C++ lines, and BE build-hygiene checks. - [x] Groovy 4.0.19: reproduce the null port before the fix; verify all four port lookups, suite setup/matrix execution, missing configuration, disabled execution and restoration after an injected query failure. The setup probe substitutes SQL execution and does not validate SQL results. - [ ] Full external ES 5/6/7/8 SQL regression run; requires the external Docker fixtures and a Doris cluster. - Behavior changed: - [x] Yes: residual ES predicates remain enforced; regression setup reads configured ports correctly. - Does this need documentation? - [x] No: existing catalog properties and scanner session variables are unchanged. ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label --- be/test/format_v2/table/es_reader_test.cpp | 210 +++++++++++++++++- .../doris/connector/es/EsQueryDslBuilder.java | 4 +- .../connector/es/EsQueryDslBuilderTest.java | 55 +++++ .../connector/es/EsScanPlanProviderTest.java | 113 +++++++++- .../es/test_es_file_scanner_v2.groovy | 59 ++++- 5 files changed, 428 insertions(+), 13 deletions(-) diff --git a/be/test/format_v2/table/es_reader_test.cpp b/be/test/format_v2/table/es_reader_test.cpp index 47ec3eca22d7c1..e143c27bf3e8d3 100644 --- a/be/test/format_v2/table/es_reader_test.cpp +++ b/be/test/format_v2/table/es_reader_test.cpp @@ -21,6 +21,8 @@ #include #include +#include +#include #include #include @@ -111,6 +113,28 @@ EsScrollStreamFactory factory(const std::shared_ptr& state) { return std::make_unique(state); }; } + +class ScrollHttpHandler final : public HttpHandler { +public: + void handle(HttpRequest* req) override { + if (req->method() == DELETE) { + EXPECT_EQ(req->get_request_body(), R"({"scroll_id":"scroll-token"})"); + ++clears; + HttpChannel::send_reply(req, HttpStatus::OK, R"({"succeeded":true})"); + } else if (searches.fetch_add(1) == 0) { + HttpChannel::send_reply(req, HttpStatus::OK, + page(empty ? "" : R"({"_source":{"id":7}})")); + } else { + HttpChannel::send_reply(req, scroll_status, R"({"error":"scroll failed"})"); + } + } + + bool empty = false; + HttpStatus scroll_status = HttpStatus::NOT_FOUND; + std::atomic searches {0}; + std::atomic clears {0}; +}; + class EsV2ReaderTest : public testing::Test { protected: void SetUp() override { @@ -130,13 +154,30 @@ class EsV2ReaderTest : public testing::Test { {"_id":"b","_source":{"id":2,"name":null,"values":[]}})"), page(R"({"_id":"c","_source":{"id":3,"name":"three","values":null}})")}; } - std::unique_ptr reader() { + std::unique_ptr reader(bool real_http = false) { auto props = std::make_shared(); auto file = std::make_unique(); file->path = "es://example/0"; + auto result = std::make_unique( + props, file, _io, &_profile, _range, _params, _slots, + real_http ? EsScrollStreamFactory {} : factory(_responses)); + EXPECT_TRUE(result->init(&_state).ok()); + return result; + } + std::unique_ptr http_reader(EvHttpServer& server, HttpHandler* handler) { + EXPECT_TRUE(server.register_handler(POST, "/example/_search", handler)); + EXPECT_TRUE(server.register_handler(POST, "/_search/scroll", handler)); + EXPECT_TRUE(server.register_handler(DELETE, "/_search/scroll", handler)); + server.start(); + EXPECT_GT(server.get_real_port(), 0); + _range.table_format_params.es_params["host_port"] = + "http://127.0.0.1:" + std::to_string(server.get_real_port()); + auto props = std::make_shared(); + auto file = std::make_unique(); auto result = std::make_unique(props, file, _io, &_profile, _range, _params, - _slots, factory(_responses)); + _slots); EXPECT_TRUE(result->init(&_state).ok()); + result->set_batch_size(1); return result; } std::shared_ptr request(std::initializer_list ids) { @@ -537,4 +578,169 @@ TEST_F(EsV2ReaderTest, FullyFilteredBatchesKeepScanningUntilEof) { EXPECT_EQ(_responses->closes, 1); } +TEST_F(EsV2ReaderTest, SourceOnlyFieldsDisableDocvalues) { + for (int source_column : {1, 2}) { + _responses = std::make_shared(); + _responses->pages = {page(R"({"_id":"a","_source":{"id":1,"name":"one"}})")}; + _params.es_properties["doc_values_mode"] = "1"; + _params.__set_es_docvalue_context({{"id", "id"}, {"_id", "_id"}}); + auto scan = reader(); + auto projection = request({0, source_column}); + ASSERT_TRUE(scan->open(projection).ok()); + EXPECT_FALSE(_responses->docvalues); + rapidjson::Document query; + query.Parse(_responses->properties.at("query").c_str()); + ASSERT_FALSE(query.HasParseError()); + EXPECT_TRUE(query.HasMember("_source")); + EXPECT_FALSE(query.HasMember("docvalue_fields")); + auto output = block(*scan, *projection); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + EXPECT_EQ(int_at(output, 0, 0), 1); + EXPECT_EQ(string_at(output, 1, 0), source_column == 1 ? "one" : "a"); + ASSERT_TRUE(scan->close().ok()); + EXPECT_EQ(_responses->closes, 1); + } +} + +TEST_F(EsV2ReaderTest, DocvaluesPreserveMissingFieldsAndEmptyArrays) { + _params.es_properties["doc_values_mode"] = "1"; + _params.__set_es_docvalue_context( + {{"id", "id"}, {"name", "name.keyword"}, {"values", "values"}}); + _responses->pages = {page(R"({"fields":{"id":[1],"name.keyword":["one"],"values":[1,3]}}, + {"fields":{"id":[2],"values":[]}}, + {"fields":{"id":[3],"name.keyword":null,"values":null}})")}; + auto scan = reader(); + auto projection = request({1, 3, 0}); + ASSERT_TRUE(scan->open(projection).ok()); + EXPECT_TRUE(_responses->docvalues); + auto output = block(*scan, *projection); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok()); + ASSERT_EQ(rows, 3); + for (const auto& column : output) { + EXPECT_EQ(column.column->size(), rows); + } + EXPECT_EQ(string_at(output, 0, 0), "one"); + EXPECT_TRUE(output.get_by_position(0).column->is_null_at(1)); + EXPECT_TRUE(output.get_by_position(0).column->is_null_at(2)); + EXPECT_FALSE(output.get_by_position(1).column->is_null_at(1)); + EXPECT_TRUE(output.get_by_position(1).column->is_null_at(2)); + const auto& values = assert_cast( + assert_cast(*output.get_by_position(1).column) + .get_nested_column()); + EXPECT_EQ(values.get_offsets()[0], 2); + EXPECT_EQ(values.get_offsets()[1], 2); + EXPECT_EQ(values.get_offsets()[2], 2); + EXPECT_EQ(int_at(output, 2, 2), 3); +} + +TEST_F(EsV2ReaderTest, RealHttpEarlyCloseClearsInitialScroll) { + ScrollHttpHandler handler; + EvHttpServer server {0}; + auto scan = http_reader(server, &handler); + ASSERT_TRUE(scan->open(request({0})).ok()); + ASSERT_TRUE(scan->close().ok()); + scan.reset(); + EXPECT_EQ(handler.searches.load(), 1); + EXPECT_EQ(handler.clears.load(), 1); +} + +TEST_F(EsV2ReaderTest, RealHttpEmptyResultClearsScroll) { + ScrollHttpHandler handler; + handler.empty = true; + EvHttpServer server {0}; + auto scan = http_reader(server, &handler); + auto projection = request({0}); + ASSERT_TRUE(scan->open(projection).ok()); + auto output = block(*scan, *projection); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok()); + EXPECT_EQ(rows, 0); + EXPECT_TRUE(eof); + ASSERT_TRUE(scan->close().ok()); + EXPECT_EQ(handler.searches.load(), 1); + EXPECT_EQ(handler.clears.load(), 1); +} + +TEST_F(EsV2ReaderTest, RealHttpScrollErrorsFailAndReleaseContext) { + for (auto status : {HttpStatus::NOT_FOUND, HttpStatus::INTERNAL_SERVER_ERROR}) { + ScrollHttpHandler handler; + handler.scroll_status = status; + EvHttpServer server {0}; + auto scan = http_reader(server, &handler); + auto projection = request({0}); + ASSERT_TRUE(scan->open(projection).ok()); + auto output = block(*scan, *projection); + size_t rows = 0; + bool eof = false; + ASSERT_TRUE(scan->get_block(&output, &rows, &eof).ok()); + ASSERT_EQ(rows, 1); + output = block(*scan, *projection); + const auto result = scan->get_block(&output, &rows, &eof); + EXPECT_FALSE(result.ok()); + EXPECT_NE(result.to_string().find(status == HttpStatus::NOT_FOUND ? "No search context" + : "scroll failed"), + std::string::npos); + ASSERT_TRUE(scan->close().ok()); + scan.reset(); + EXPECT_EQ(handler.searches.load(), 2); + EXPECT_EQ(handler.clears.load(), 1); + } +} + +// The opt-in fixture has integer id / keyword name and rows (1,"alpha"), (2,"beta"), (3,null). +TEST_F(EsV2ReaderTest, ConfiguredHttpSourceSupportsSourceAndDocvalues) { + const char* host = std::getenv("DORIS_ES_TEST_HOST"); + if (host == nullptr) { + GTEST_SKIP() << "Set DORIS_ES_TEST_HOST and DORIS_ES_TEST_INDEX for a live ES smoke test"; + } + const char* index = std::getenv("DORIS_ES_TEST_INDEX"); + ASSERT_NE(index, nullptr); + _range.table_format_params.es_params["host_port"] = host; + _range.table_format_params.es_params["index"] = index; + _range.table_format_params.es_params["shard_id"] = "-1"; + for (const auto& [property, variable] : {std::pair {"user", "DORIS_ES_TEST_USER"}, + std::pair {"password", "DORIS_ES_TEST_PASSWORD"}}) { + if (const char* value = std::getenv(variable); value != nullptr) { + _params.es_properties[property] = value; + } + } + _params.__set_es_docvalue_context({{"id", "id"}, {"name", "name"}}); + for (bool docvalues : {false, true}) { + _params.es_properties["doc_values_mode"] = docvalues ? "1" : "0"; + auto scan = reader(true); + scan->set_batch_size(1); + auto projection = request({1, 0}); + const auto opened = scan->open(projection); + ASSERT_TRUE(opened.ok()) << opened; + size_t total = 0; + std::set ids; + bool eof = false; + while (!eof) { + auto output = block(*scan, *projection); + size_t rows = 0; + const auto status = scan->get_block(&output, &rows, &eof); + ASSERT_TRUE(status.ok()) << status; + for (size_t row = 0; row < rows; ++row) { + const int32_t id = int_at(output, 1, row); + ids.insert(id); + if (id == 3) { + EXPECT_TRUE(output.get_by_position(0).column->is_null_at(row)); + } else { + EXPECT_EQ(string_at(output, 0, row), id == 1 ? "alpha" : "beta"); + } + } + total += rows; + } + EXPECT_EQ(total, 3); + EXPECT_EQ(ids, (std::set {1, 2, 3})); + ASSERT_TRUE(scan->close().ok()); + } +} + } // namespace doris::format::es diff --git a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsQueryDslBuilder.java b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsQueryDslBuilder.java index d376e345bf5038..598d88a1f7eab9 100644 --- a/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsQueryDslBuilder.java +++ b/fe/fe-connector/fe-connector-es/src/main/java/org/apache/doris/connector/es/EsQueryDslBuilder.java @@ -147,7 +147,9 @@ public static EsQueryDslResult buildQueryDslWithResult(ConnectorExpression filte List notPushDownList = new ArrayList<>(); String dsl = toEsDsl(filter, notPushDownList, fieldsContext, likePushDown, needCompatDateFields, column2typeMap); - if (dsl == null) { + // A nested AND can return a partial DSL while recording an unsupported child. OR/NOT must + // remain local in that case: e.g. NOT(a = 1 AND local_predicate) cannot become NOT(a = 1). + if (dsl == null || !notPushDownList.isEmpty()) { return new EsQueryDslResult(matchAllQuery(), Collections.singletonList(0)); } return new EsQueryDslResult(dsl, Collections.emptyList()); diff --git a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsQueryDslBuilderTest.java b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsQueryDslBuilderTest.java index 3991935601e8dc..fe379a0e4b5cf5 100644 --- a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsQueryDslBuilderTest.java +++ b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsQueryDslBuilderTest.java @@ -18,18 +18,73 @@ package org.apache.doris.connector.es; import org.apache.doris.connector.spi.ConnectorType; +import org.apache.doris.connector.spi.pushdown.ConnectorAnd; import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef; import org.apache.doris.connector.spi.pushdown.ConnectorComparison; +import org.apache.doris.connector.spi.pushdown.ConnectorExpression; +import org.apache.doris.connector.spi.pushdown.ConnectorFunctionCall; import org.apache.doris.connector.spi.pushdown.ConnectorLike; import org.apache.doris.connector.spi.pushdown.ConnectorLiteral; +import org.apache.doris.connector.spi.pushdown.ConnectorNot; +import org.apache.doris.connector.spi.pushdown.ConnectorOr; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.Collections; public class EsQueryDslBuilderTest { + private static ConnectorExpression keywordEquals(String value) { + return new ConnectorComparison(ConnectorComparison.Operator.EQ, + new ConnectorColumnRef("name", ConnectorType.of("VARCHAR")), ConnectorLiteral.ofString(value)); + } + + private static ConnectorExpression partiallyPushableAnd() { + ConnectorExpression local = new ConnectorFunctionCall("unsupported_predicate", ConnectorType.of("BOOLEAN"), + Collections.singletonList(new ConnectorColumnRef("name", ConnectorType.of("VARCHAR")))); + return new ConnectorAnd(Arrays.asList(keywordEquals("one"), local)); + } + + private static EsQueryDslResult buildWithTracking(ConnectorExpression expression) { + return EsQueryDslBuilder.buildQueryDslWithResult(expression, Collections.emptyMap(), + Collections.singletonMap("name", "keyword"), true, Collections.emptyList()); + } + + @Test + void partialAndRetainsOnlyItsUnsupportedConjunct() { + EsQueryDslResult result = buildWithTracking(partiallyPushableAnd()); + Assertions.assertEquals(Collections.singletonList(1), result.getNotPushedIndices()); + Assertions.assertEquals("{\"term\":{\"name\":\"one\"}}", result.getQueryDsl()); + } + + @Test + void orWithPartiallyPushableAndRemainsLocal() { + EsQueryDslResult result = buildWithTracking(new ConnectorOr( + Arrays.asList(partiallyPushableAnd(), keywordEquals("two")))); + Assertions.assertEquals(Collections.singletonList(0), result.getNotPushedIndices()); + Assertions.assertEquals(EsQueryDslBuilder.matchAllQuery(), result.getQueryDsl()); + } + + @Test + void notWithPartiallyPushableAndRemainsLocal() { + EsQueryDslResult result = buildWithTracking(new ConnectorNot(partiallyPushableAnd())); + Assertions.assertEquals(Collections.singletonList(0), result.getNotPushedIndices()); + Assertions.assertEquals(EsQueryDslBuilder.matchAllQuery(), result.getQueryDsl()); + } + + @Test + void fullyPushableOrAndNotRemainPushed() { + for (ConnectorExpression expression : Arrays.asList( + new ConnectorOr(Arrays.asList(keywordEquals("one"), keywordEquals("two"))), + new ConnectorNot(keywordEquals("one")))) { + EsQueryDslResult result = buildWithTracking(expression); + Assertions.assertTrue(result.getNotPushedIndices().isEmpty()); + Assertions.assertNotEquals(EsQueryDslBuilder.matchAllQuery(), result.getQueryDsl()); + } + } + @Test public void testRegexpProducesRegexpQuery() { // SQL: WHERE name REGEXP '^test.*' diff --git a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsScanPlanProviderTest.java b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsScanPlanProviderTest.java index a316f63ce05a03..1f43af31948573 100644 --- a/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsScanPlanProviderTest.java +++ b/fe/fe-connector/fe-connector-es/src/test/java/org/apache/doris/connector/es/EsScanPlanProviderTest.java @@ -21,7 +21,9 @@ import org.apache.doris.connector.spi.ConnectorContractValidator; import org.apache.doris.connector.spi.ConnectorSession; import org.apache.doris.connector.spi.ConnectorStatementScope; +import org.apache.doris.connector.spi.handle.ConnectorColumnHandle; import org.apache.doris.connector.spi.handle.NamedColumnHandle; +import org.apache.doris.connector.spi.scan.ConnectorScanRange; import org.apache.doris.connector.spi.scan.ConnectorScanRequest; import org.apache.doris.connector.spi.scan.ScanNodePropertyKeys; import org.apache.doris.thrift.TFileScanRangeParams; @@ -29,9 +31,12 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; @@ -82,9 +87,15 @@ public Map getHttpNodes() { */ private static final class TestSession implements ConnectorSession { private final ConnectorStatementScope scope; + private final Map sessionProperties; TestSession(ConnectorStatementScope scope) { + this(scope, Collections.emptyMap()); + } + + TestSession(ConnectorStatementScope scope, Map sessionProperties) { this.scope = scope; + this.sessionProperties = sessionProperties; } @Override @@ -129,7 +140,7 @@ public Map getCatalogProperties() { @Override public Map getSessionProperties() { - return Collections.emptyMap(); + return sessionProperties; } @Override @@ -459,5 +470,105 @@ public void beFileCacheAdmissionDoesNotApplyToEs() { .supportsFileCache()); } + @Test + void docvalueEligibilityHonorsCatalogFieldLimitAndSourceOnlyColumns() { + List one = Collections.singletonList(new NamedColumnHandle("a")); + List two = Arrays.asList(new NamedColumnHandle("a"), new NamedColumnHandle("b")); + List id = Arrays.asList(new NamedColumnHandle("a"), new NamedColumnHandle("_id")); + Map properties = minimalPropertyMap(); + properties.put(EsCatalogProperties.MAX_DOCVALUE_FIELDS, "1"); + EsScanPlanProvider provider = new EsScanPlanProvider( + new CountingRestClient(), EsCatalogProperties.of(properties)); + EsTableHandle handle = new EsTableHandle("test_index"); + Assertions.assertEquals("1", provider.getScanNodePropertiesResult( + EMPTY_SESSION, handle, one, Optional.empty()).getProperties().get("doc_values_mode")); + Assertions.assertEquals("0", provider.getScanNodePropertiesResult( + EMPTY_SESSION, handle, two, Optional.empty()).getProperties().get("doc_values_mode")); + + properties.put(EsCatalogProperties.MAX_DOCVALUE_FIELDS, "2"); + provider = new EsScanPlanProvider(new CountingRestClient(), EsCatalogProperties.of(properties)); + Assertions.assertEquals("1", provider.getScanNodePropertiesResult( + EMPTY_SESSION, handle, two, Optional.empty()).getProperties().get("doc_values_mode")); + Assertions.assertEquals("0", provider.getScanNodePropertiesResult( + EMPTY_SESSION, handle, id, Optional.empty()).getProperties().get("doc_values_mode")); + + properties.put(EsCatalogProperties.DOC_VALUE_SCAN, "false"); + provider = new EsScanPlanProvider(new CountingRestClient(), EsCatalogProperties.of(properties)); + Assertions.assertEquals("0", provider.getScanNodePropertiesResult( + EMPTY_SESSION, handle, one, Optional.empty()).getProperties().get("doc_values_mode")); + } + + @Test + void keywordAliasesAndSessionBatchSizeSurviveThriftPopulation() { + CountingRestClient client = new CountingRestClient() { + @Override + public String getMapping(String indexName) { + return "{\"" + indexName + "\":{\"mappings\":{\"properties\":{" + + "\"a\":{\"type\":\"text\",\"fields\":{\"keyword\":{\"type\":\"keyword\"}}}," + + "\"b\":{\"type\":\"keyword\",\"doc_values\":false}}}}}"; + } + }; + EsScanPlanProvider provider = new EsScanPlanProvider(client, minimalProps()); + ConnectorSession session = new TestSession(ConnectorStatementScope.NONE, + Collections.singletonMap("batch_size", "64")); + EsTableHandle handle = new EsTableHandle("test_index"); + Map properties = provider.getScanNodePropertiesResult(session, handle, + Collections.singletonList(new NamedColumnHandle("a")), Optional.empty()).getProperties(); + Assertions.assertEquals("1", properties.get("doc_values_mode")); + Assertions.assertEquals("64", properties.get("es.batch_size")); + TFileScanRangeParams params = new TFileScanRangeParams(); + provider.populateScanLevelParams(params, properties); + Assertions.assertEquals(Collections.singletonMap("a", "a.keyword"), params.getEsDocvalueContext()); + Assertions.assertEquals(Collections.singletonMap("a", "a.keyword"), params.getEsFieldsContext()); + Assertions.assertEquals("{\"match_all\":{}}", params.getEsProperties().get("query_dsl")); + Assertions.assertFalse(params.getEsProperties().containsKey("es.batch_size")); + + Map sourceOnly = provider.getScanNodePropertiesResult(session, handle, + Arrays.asList(new NamedColumnHandle("a"), new NamedColumnHandle("b")), + Optional.empty()).getProperties(); + Assertions.assertEquals("0", sourceOnly.get("doc_values_mode")); + } + + @Test + void parallelAndWholeIndexPlansPreserveShardRouting() { + CountingRestClient client = new CountingRestClient() { + @Override + public EsShardPartitions searchShards(String indexName) { + EsShardPartitions partitions = new EsShardPartitions(indexName); + for (int shard : new int[]{0, 1}) { + partitions.addShardRouting(shard, Collections.singletonList( + new EsShardRouting(indexName, shard, true, "node1"))); + } + return partitions; + } + }; + EsScanPlanProvider provider = new EsScanPlanProvider(client, minimalProps()); + ConnectorScanRequest request = ConnectorScanRequest.builder( + new EsTableHandle("test_index"), Collections.emptyList()).build(); + List parallel = provider.planScan(new TestSession(ConnectorStatementScope.NONE, + Collections.singletonMap("enable_es_parallel_scroll", "true")), request); + Assertions.assertEquals(2, parallel.size()); + Assertions.assertEquals(java.util.Set.of("0", "1"), parallel.stream() + .map(range -> range.getProperties().get("shard_id")).collect(java.util.stream.Collectors.toSet())); + List serial = provider.planScan(new TestSession(ConnectorStatementScope.NONE, + Collections.singletonMap("enable_es_parallel_scroll", "false")), request); + Assertions.assertEquals(1, serial.size()); + Assertions.assertEquals("-1", serial.get(0).getProperties().get("shard_id")); + Assertions.assertEquals("localhost:9200", serial.get(0).getProperties().get("es_hosts")); + Assertions.assertEquals("test_index", serial.get(0).getProperties().get("index")); + } + + @Test + void earlyStopRequiresValidPositiveNumbersAndExplicitPushdownTracking() { + Assertions.assertEquals("64", populateWith(pushdownProps("64", "64", "true")) + .getEsProperties().get("limit")); + for (String[] values : new String[][]{ + {"0", "64", "true"}, {"-1", "64", "true"}, {"65", "64", "true"}, + {"1", "0", "true"}, {"1", "-1", "true"}, {"invalid", "64", "true"}, + {"1", "invalid", "true"}, {"1", "64", null}, {"1", "64", "false"}}) { + Assertions.assertFalse(populateWith(pushdownProps(values[0], values[1], values[2])) + .getEsProperties().containsKey("limit"), Arrays.toString(values)); + } + } } diff --git a/regression-test/suites/external_table_p2/es/test_es_file_scanner_v2.groovy b/regression-test/suites/external_table_p2/es/test_es_file_scanner_v2.groovy index e23dce31b8577d..cd1a77af1f50e5 100644 --- a/regression-test/suites/external_table_p2/es/test_es_file_scanner_v2.groovy +++ b/regression-test/suites/external_table_p2/es/test_es_file_scanner_v2.groovy @@ -23,11 +23,12 @@ suite("test_es_file_scanner_v2", "p2,external,es,external_docker,external_docker String externalEnvIp = context.config.otherConfigs.get("externalEnvIp") def originalV2 = sql("show variables like 'enable_file_scanner_v2'")[0][1] def originalBatchSize = sql("show variables like 'batch_size'")[0][1] + def originalParallelScroll = sql("show variables like 'enable_es_parallel_scroll'")[0][1] try { - // Small batches force scroll pagination and repeated output block production. - sql "set batch_size = 1" [5, 6, 7, 8].each { version -> - String port = context.config.otherConfigs.get("es_${version}_port") + // Properties.get(Object) does not coerce GString keys to the stored String keys. + String port = context.config.otherConfigs.get("es_${version}_port".toString()) + assertTrue(port != null && !port.trim().isEmpty(), "Missing ES ${version} port configuration") String timeIndex = version <= 6 ? "test2_20220808" : "test2" [false, true].each { docvalues -> String catalog = "test_es_v2_${version}_${docvalues}" @@ -45,28 +46,68 @@ suite("test_es_file_scanner_v2", "p2,external,es,external_docker,external_docker "select test2, test1, test6 from test1 order by test1, test2, test6", "select _id, test1 from test1 order by _id", "select c_bool, c_integer, c_keyword, c_person from test1 order by test1", + "select c_byte, c_short, c_long, c_float, c_half_float, c_double, c_scaled_float from test1 order by test1", + "select c_date, c_datetime, c_text, c_ip, c_user from test1 order by test1", "select test1 from ${timeIndex} where substring(test2, 2) = 'ext2' order by test1", + "select test1 from ${timeIndex} where test4 >= '2022-08-08 00:00:00' and test4 < '2022-08-09 00:00:00' order by test1", + "select test1 from test1 where length(test2) > 4 order by test1", + "select test1 from test1 where length(test2) > 1000 order by test1", + "select test1 from test1 where test1 in ('string1', 'string3') order by test1", + "select test1 from test1 where test1 not in ('string1', 'string3') order by test1", + "select test1 from test1 where test1 = 'string1' or length(test2) > 5 order by test1", "select test1 from test1 where test1 like 'string%' order by test1", + """select test1 from test1 where esquery(test2, '{"match":{"test2":"text#1"}}') order by test1""", "select count(*), count(message) from test1", + "select count(*) from test1 where length(test2) > 4", "select test1 from test1 where message is null order by test1", + "select test1 from test1 where message is not null order by test1", "select test1 from test1 where test1 = 'missing-value'", + "select count(*) from test1 where test1 = 'missing-value'", "select test1 from test1 order by test1 limit 2", + "select test1 from test1 where length(test2) > 4 order by test1 limit 2", + // Unordered LIMIT may choose different documents on different BEs; compare its cardinality. + "select count(*) from (select * from test1 limit 1) t", "select count(*) from (select * from test1 limit 2) t", "select * from composite_type_array order by name" ] + if (version >= 7) { + queries.add("select id, name, tags, emails, single_value, description from test_keyword_array order by id") + } + if (version == 7) { + queries.add("select extra from test_flatten order by extra") + } // Compare against the existing scanner on the same index and ES version instead // of embedding fixture results that can vary across the external Docker images. - queries.each { query -> - sql "set enable_file_scanner_v2 = false" - def expected = sql(query) - sql "set enable_file_scanner_v2 = true" - def actual = sql(query) - assertEquals(expected, actual, "ES ${version}, docvalues=${docvalues}: ${query}") + [false, true].each { parallelScroll -> + sql "set enable_es_parallel_scroll = ${parallelScroll}" + // Exercise scroll pagination and the one-request LIMIT path at their batch boundaries. + [1, 1024].each { batchSize -> + sql "set batch_size = ${batchSize}" + queries.each { query -> + sql "set enable_file_scanner_v2 = false" + def expected = sql(query) + sql "set enable_file_scanner_v2 = true" + def actual = sql(query) + assertEquals(expected, actual, + "ES ${version}, docvalues=${docvalues}, parallel=${parallelScroll}, batch=${batchSize}: ${query}") + } + // Both scanners share FE pushdown, so use an independent predicate as the oracle. + def expectedOr = sql("select test1 from test1 where test1 = 'string2' order by test1") + [false, true].each { scannerV2 -> + sql "set enable_file_scanner_v2 = ${scannerV2}" + def actualOr = sql("""select test1 from test1 + where (test1 = 'string1' and length(test2) < 0) or test1 = 'string2' + order by test1""") + assertEquals(expectedOr, actualOr, + "ES ${version}, docvalues=${docvalues}, parallel=${parallelScroll}, batch=${batchSize}, v2=${scannerV2}: nested residual OR") + } + } } } } } finally { sql "set enable_file_scanner_v2 = ${originalV2}" sql "set batch_size = ${originalBatchSize}" + sql "set enable_es_parallel_scroll = ${originalParallelScroll}" } }