github-actions[bot] commented on code in PR #68883:
URL: https://github.com/apache/doris/pull/68883#discussion_r4237385154
##########
be/src/core/data_type_serde/data_type_variant_v2_serde.cpp:
##########
@@ -715,6 +831,9 @@ Status DataTypeVariantV2SerDe::write_column_to_arrow(const
IColumn& column, cons
options.timezone = &ctz;
const size_t first = checked_row(start);
const size_t last = checked_row(end);
+ if (array_builder->type()->id() == arrow::Type::STRUCT) {
Review Comment:
[P1] Preserve the value-first Variant struct writer. This branch captures
every struct builder, but `write_parquet_variant_arrow` rejects the existing
`struct<value: binary, metadata: binary>` layout before the following
`write_binary_variant_arrow` branch can run. Paimon's Variant writer uses that
value-first schema and delegates here, so Paimon Variant writes now fail; the
unchanged binary-struct SerDe tests would fail too. Dispatch by the validated
child field order to keep both representations reachable.
##########
be/src/exec/operator/memory_scratch_sink_operator.cpp:
##########
@@ -102,8 +102,7 @@ Status MemoryScratchSinkOperatorX::sink_impl(RuntimeState*
state, Block* input_b
RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(
local_state._output_vexpr_ctxs, *input_block, &block));
}
- DorisArrowBlockConvertor converter(block, state->timezone(), _timezone_obj,
- /*datetime_naive=*/true);
+ ArrowFlightArrowBlockConvertor converter(block, state->timezone(),
_timezone_obj);
Review Comment:
[P1] Keep the external scanner's Variant wire type compatible.
`TableQueryPlanAction` installs this memory scratch sink for `open_scanner`,
and `get_next` sends its RecordBatch directly to scanner clients. Switching to
the Flight converter changes a selected Variant V2 column from UTF8 to an
extension-backed STRUCT without any scanner capability flag; the in-tree
`DorisReaderExample` accepts VARCHAR but rejects STRUCT. Keep this sink on the
ordinary converter unless the separate scanner protocol negotiates native
Variant.
##########
fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java:
##########
@@ -75,6 +84,56 @@ public FlightSqlSchemaHelper(ConnectContext context) {
private static final byte[] EMPTY_SERIALIZED_SCHEMA =
getSerializedSchema(Collections.emptyList());
+ static Field withDorisTypeMetadata(Field field, Type type) {
+ if (type.isVariantType()) {
+ if (field.getMetadata() == null
+ ||
!"arrow.parquet.variant".equals(field.getMetadata().get("ARROW:extension:name")))
{
+ throw CallStatus.UNIMPLEMENTED.withDescription(
Review Comment:
[P2] Preserve the unsupported Variant status at the Flight boundary. An
older BE can return a UTF8 schema for a planned Variant, causing this branch to
throw UNIMPLEMENTED. `fetchArrowFlightSchema` wraps that exception, then
`executeQueryStatement` passes it to `queryFailure`, which returns INTERNAL to
the client. Pass the capability error through the fetch and producer boundaries
so clients receive the actionable UNIMPLEMENTED status.
##########
regression-test/suites/arrow_flight_sql_p0/test_flight_native_variant.groovy:
##########
@@ -0,0 +1,174 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.apache.arrow.driver.jdbc.shaded.com.google.protobuf.Any
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CallOptions
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.FlightClient
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.Location
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.sql.FlightSqlClient
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.sql.impl.FlightSql
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.memory.RootAllocator
+
+import java.util.concurrent.TimeUnit
+
+suite("test_flight_native_variant", "arrow_flight_sql") {
+ def frontend = jdbc_sql_return_maparray("SHOW FRONTENDS").find {
+ it.IsMaster.toString().equalsIgnoreCase("true") &&
it.Alive.toString().equalsIgnoreCase("true")
+ }
+ assertNotNull(frontend)
+ assertTrue(frontend.ArrowFlightSqlPort.toString().toInteger() > 0)
+ def database = jdbc_sql("SELECT DATABASE()")[0][0]
+ def table = "${database}.flight_native_variant_input"
+ def allocator = new RootAllocator(Long.MAX_VALUE)
+ def feClient = FlightClient.builder(allocator,
+ Location.forGrpcInsecure(frontend.Host.toString(),
frontend.ArrowFlightSqlPort.toString().toInteger())).build()
+ def client = new FlightSqlClient(feClient)
+ def auth
+ def read = { String query, Closure inspect, boolean parallel = false, int
resultBackendCount = 1 ->
+ int count = 0
+ def info = client.execute(query, auth)
+ assertFalse(info.endpoints.isEmpty())
+ // Multiple buckets alone do not prove coverage of native output from
different result BEs.
+ if (parallel) {
+ def resultAddresses = info.endpoints.collect { endpoint ->
+ def fields =
Any.parseFrom(endpoint.ticket.bytes).unpack(FlightSql.TicketStatementQuery.class)
+ .statementHandle.toStringUtf8().split("&")
+ "${fields[1]}:${fields[2]}".toString()
+ }
+ assertEquals(info.endpoints.size(),
resultAddresses.toSet().size(), "Duplicate result backends")
+ if (resultBackendCount > 1) {
+ assertTrue(info.endpoints.size() > 1, "Expected multiple
native Variant result backends")
+ }
+ }
+ info.endpoints.each { endpoint ->
+ FlightClient.builder(allocator,
endpoint.locations[0]).build().withCloseable { beClient ->
+ beClient.getStream(endpoint.ticket, auth,
CallOptions.timeout(30, TimeUnit.SECONDS)).withCloseable { stream ->
+ assertEquals(info.schema, stream.schema, "Result endpoints
must share the published schema")
+ while (stream.next()) {
+ inspect(stream.root)
+ count += stream.root.rowCount
+ }
+ }
+ }
+ }
+ count
+ }
+ def executeSetting = { String query -> read(query, { root -> }) }
+ try {
+ auth =
feClient.authenticateBasicToken(context.config.otherConfigs.get("extArrowFlightSqlUser"),
+
context.config.otherConfigs.get("extArrowFlightSqlPassword")).get()
+ executeSetting("SET enable_sql_cache=false")
+ executeSetting("SET enable_nereids_distribute_planner=true")
+ executeSetting("SET parallel_pipeline_task_num=8")
+ jdbc_sql("DROP TABLE IF EXISTS ${table}")
+ jdbc_sql("""CREATE TABLE ${table} (id INT, v VARIANT)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 60
+ PROPERTIES("replication_num"="1")""")
+ jdbc_sql("""INSERT INTO ${table}
+ SELECT number + 1, parse_to_variant(CASE number % 4
+ WHEN 0 THEN '42' WHEN 1 THEN '"text"'
+ WHEN 2 THEN '{"a":[1,null,"x"]}' ELSE NULL END)
+ FROM numbers("number"="60")""")
+ def resultBackendCount = jdbc_sql_return_maparray("SHOW TABLETS FROM
${table}")
+ .collect { it.BackendId }.unique().size()
+ [false, true].each { parallel ->
+ executeSetting("SET enable_parallel_result_sink=${parallel}")
+ // Clients needing JSON strings request text through an explicit
SQL conversion.
+ assertEquals(60, read("SELECT CAST(v AS STRING) AS text_value FROM
${table}", { root ->
+ assertEquals("Utf8", root.getVector(0).field.type.toString())
+ }))
+ def seen = []
+ assertEquals(60, read("SELECT id, v FROM ${table}", { root ->
+ def vector = root.getVector(1)
+ def field = vector.field
+ assertEquals("arrow.parquet.variant",
field.metadata.get("ARROW:extension:name"))
+ assertEquals("Struct", field.type.toString())
+ assertEquals(["metadata", "value"], field.children.collect {
it.name })
+ field.children.each { child ->
+ assertFalse(child.nullable)
+ assertEquals("Binary", child.type.toString())
+ }
+ for (int i = 0; i < root.rowCount; i++) {
+ int id = root.getVector(0).get(i)
+ seen.add(id)
+ assertEquals(id % 4 == 0, vector.isNull(i))
+ if (id % 4 != 0) {
+ assertTrue(vector.getChild("metadata").get(i).length >
0)
+ assertTrue(vector.getChild("value").get(i).length > 0)
+ if (id % 4 == 1) {
+ assertEquals([12, 42],
vector.getChild("value").get(i).collect { it & 0xff })
+ }
+ }
+ }
+ }, parallel, resultBackendCount))
+ assertEquals((1..60).toList(), seen.sort())
+ def scannedColumns = "v, ARRAY(v) AS a, MAP('key', v) AS m,
STRUCT(v) AS s"
+ // Master obtains result schemas through execution; cover nested
and empty results there.
+ ["SELECT CAST(42 AS VARIANT) AS v",
+ "SELECT ${scannedColumns} FROM ${table} WHERE id = 1",
+ "SELECT ${scannedColumns} FROM ${table} WHERE id <
0"].eachWithIndex { query, index ->
+ assertEquals(index == 2 ? 0 : 1, read(query.toString(), { root
->
+ def leaves = [root.schema.fields[0]]
+ if (index > 0) {
+ leaves.add(root.schema.fields[1].children[0])
+
leaves.add(root.schema.fields[2].children[0].children[1])
+ leaves.add(root.schema.fields[3].children[0])
+ }
+ leaves.each { field ->
+ assertEquals("Struct", field.type.toString())
+ assertEquals("arrow.parquet.variant",
field.metadata.get("ARROW:extension:name"))
+ }
+ }))
+ }
+ }
+ // Each row owns its wire dictionary; unrelated keys must not multiply
Arrow metadata.
+ def keyExpression = "CONCAT(REPEAT('k', 244), LPAD(CAST(number AS
STRING), 6, '0'))"
+ def variantExpression = """parse_to_variant(CONCAT('{"',
${keyExpression}, '":', CAST(number AS STRING), '}'))"""
+ def metadataRows = []
+ assertEquals(128, read("SELECT number AS id, ${variantExpression} AS v
FROM numbers(\"number\"=\"128\")", { root ->
+ def variant = root.getVector(1)
+ assertEquals("arrow.parquet.variant",
variant.field.metadata.get("ARROW:extension:name"))
+ for (int i = 0; i < root.rowCount; ++i) {
+ metadataRows.add(root.getVector(0).get(i).intValue())
+ assertTrue(variant.getChild("metadata").get(i).length < 270,
+ "A row must not repeat the batch dictionary")
+ }
+ }))
+ assertEquals((0..<128).toList(), metadataRows.sort())
+ // A folded constant must use the same wire representation as a
scanned Variant column.
+ assertEquals(1, read("SELECT parse_to_variant('42') AS v", { root ->
+ assertEquals("arrow.parquet.variant",
root.getVector(0).field.metadata.get("ARROW:extension:name"))
+ }))
+ assertEquals(0, read("SELECT parse_to_variant(CAST(id AS STRING)) AS v
FROM ${table} WHERE id < 0", { root -> }))
+ } finally {
+ try {
+ if (auth != null) {
+ feClient.closeSession(new
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CloseSessionRequest(),
auth)
+ }
+ } finally {
+ try {
+ client.close()
+ } finally {
+ try {
+ allocator.close()
+ } finally {
+ jdbc_sql("DROP TABLE IF EXISTS ${table}")
Review Comment:
[P3] Leave the table available for regression debugging. This `finally` drop
removes `flight_native_variant_input` even when an assertion fails. The suite
already drops it before creation, and the repository testing standard requires
retaining test tables afterward. Keep the client and allocator cleanup, but
remove this final table drop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
