This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f109e130f77 [fix](arrow-flight) Support table type metadata and
filters (#67899)
f109e130f77 is described below
commit f109e130f77c672f763a5082c9be856380bd32ae
Author: hutiefang76 <[email protected]>
AuthorDate: Tue Nov 3 02:21:21 2026 -0800
[fix](arrow-flight) Support table type metadata and filters (#67899)
### What problem does this PR solve?
Related to #67578 (partial A3).
Flight SQL clients can obtain FlightInfo for GetTableTypes, but reading
its ticket fails with UNIMPLEMENTED. GetTables also forwards only the
first table_types entry to a service that treats only VIEW as a filter:
[VIEW, BASE TABLE] loses base tables, while [BASE TABLE] or an unknown
type can return excluded rows.
Implement the type stream using the same Doris names as GetTables (BASE
TABLE, SYSTEM VIEW and VIEW), sorted according to Flight SQL. Match the
complete filter against the existing authorized listing before
collecting optional schemas. Request only name/type metadata to avoid
unused table statistics, and retain the session guard and Arrow buffer
cleanup on failed sends.
A type is the TABLE_TYPE that information_schema.tables reports for the
object: table streams are VIEW, and external catalogs report their
tables and views as BASE TABLE. The service lists every authorized
object and the filter runs on that listing. Requested names match
case-insensitively, and TABLE and SYSTEM TABLE, which Flight SQL calls
commonly supported, select BASE TABLE and SYSTEM VIEW: the ADBC Flight
SQL driver asks for `table` (Doris's own ADBC catalog lists a Doris
source with `getObjects(..., {"table"})`), and JDBC tools ask for
`TABLE`. Any other name matches only a type of that name, so UNKNOWN
selects nothing.
Because the filter now runs on the full listing,
`FrontendServiceImpl.listTableStatus` refreshes a cloud follower's
partition versions from the meta service only when UPDATE_TIME is
requested, the one column that reads them. GetTables, which asks for
names and types, never waits on those RPCs or is cut short when they
fail, and information_schema.tables scans that do not read UPDATE_TIME
skip the refresh too.
Co-authored-by: morningman <[email protected]>
Co-authored-by: Gabriel <[email protected]>
---
.../doris/arrowflight/DorisFlightSqlProducer.java | 13 +-
.../doris/arrowflight/FlightSqlSchemaHelper.java | 66 +++++++-
.../apache/doris/service/FrontendServiceImpl.java | 7 +-
.../FlightSqlSchemaHelperTableTypesTest.java | 177 +++++++++++++++++++++
.../doris/arrowflight/FlightSqlTableTypesTest.java | 151 ++++++++++++++++++
.../doris/service/FrontendServiceImplTest.java | 57 +++++++
.../test_table_types_over_arrow_flight.groovy | 141 ++++++++++++++++
7 files changed, 601 insertions(+), 11 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/DorisFlightSqlProducer.java
b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/DorisFlightSqlProducer.java
index 5a7c1dd1f8b..a90c86f279f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/DorisFlightSqlProducer.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/DorisFlightSqlProducer.java
@@ -619,7 +619,16 @@ public class DorisFlightSqlProducer implements
FlightSqlProducer, AutoCloseable
@Override
public void getStreamTableTypes(final CallContext context, final
ServerStreamListener listener) {
- throw CallStatus.UNIMPLEMENTED.withDescription("getStreamTableTypes
unimplemented").toRuntimeException();
+ streamMetadata(context, listener, connectContext -> {
+ try (VectorSchemaRoot vectorSchemaRoot = VectorSchemaRoot.create(
+ Schemas.GET_TABLE_TYPES_SCHEMA, rootAllocator)) {
+ listener.start(vectorSchemaRoot);
+ vectorSchemaRoot.allocateNew();
+ FlightSqlSchemaHelper.getTableTypes(vectorSchemaRoot);
+ listener.putNext();
+ listener.completed();
+ }
+ });
}
@Override
@@ -761,7 +770,7 @@ public class DorisFlightSqlProducer implements
FlightSqlProducer, AutoCloseable
void send(ConnectContext connectContext) throws Exception;
}
- // Answers a metadata request (catalogs, schemas, tables) from the
session's context, as one
+ // Answers a metadata request (catalogs, schemas, tables, table types)
from the session's context, as one
// command of that session.
private void streamMetadata(final CallContext context, final
ServerStreamListener listener,
MetadataStream stream) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java
b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java
index 336c3dc7a36..f7db5af80ba 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/arrowflight/FlightSqlSchemaHelper.java
@@ -22,6 +22,7 @@ import org.apache.doris.catalog.ArrayType;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MapType;
import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.catalog.Type;
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.qe.ConnectContext;
@@ -64,18 +65,27 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
+import java.util.stream.Collectors;
public class FlightSqlSchemaHelper {
private static final Logger LOG =
LogManager.getLogger(FlightSqlSchemaHelper.class);
+ private static final List<String> TABLE_TYPES =
Arrays.stream(TableType.values())
+ .map(TableType::toMysqlType)
+ .filter(Objects::nonNull)
+ .distinct()
+ .sorted()
+ .collect(Collectors.toList());
private final ConnectContext ctx;
private final FrontendServiceImpl impl;
private boolean includeSchema;
private String catalogFilterPattern = null;
private String dbSchemaFilterPattern = null;
private String tableNameFilterPattern = null;
- private List<String> tableTypesList = null;
+ private Set<String> tableTypeFilter = null;
public FlightSqlSchemaHelper(ConnectContext context) {
ctx = context;
@@ -161,7 +171,28 @@ public class FlightSqlSchemaHelper {
catalogFilterPattern = command.hasCatalog() ? command.getCatalog() :
"internal";
dbSchemaFilterPattern = command.hasDbSchemaFilterPattern() ?
command.getDbSchemaFilterPattern() : null;
tableNameFilterPattern = command.hasTableNameFilterPattern() ?
command.getTableNameFilterPattern() : null;
- tableTypesList = command.getTableTypesList().isEmpty() ? null :
command.getTableTypesList();
+ tableTypeFilter = command.getTableTypesList().isEmpty() ? null
+ :
command.getTableTypesList().stream().map(FlightSqlSchemaHelper::toDorisTableType)
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * Maps a type a GetTables request filters on to the name GetTableTypes
advertises. Flight SQL leaves the
+ * names to the server but calls TABLE, VIEW and SYSTEM TABLE the commonly
supported ones, and clients send
+ * those in either case: the ADBC Flight SQL driver asks for "table", JDBC
tools for "TABLE". So a type
+ * matches case-insensitively, and TABLE and SYSTEM TABLE select BASE
TABLE and SYSTEM VIEW. Any other name
+ * matches only a type of that name.
+ */
+ private static String toDorisTableType(String requestedType) {
+ String type = requestedType.toUpperCase(Locale.ROOT);
+ switch (type) {
+ case "TABLE":
+ return "BASE TABLE";
+ case "SYSTEM TABLE":
+ return "SYSTEM VIEW";
+ default:
+ return type;
+ }
}
/**
@@ -199,9 +230,8 @@ public class FlightSqlSchemaHelper {
if (tableNameFilterPattern != null) {
getTablesParams.setPattern(tableNameFilterPattern);
}
- if (tableTypesList != null) {
- getTablesParams.setType(tableTypesList.get(0)); // currently only
one type is supported.
- }
+ // Flight SQL only consumes names and types, not the potentially
expensive table statistics.
+ getTablesParams.setRequiredColumns(Set.of("TABLE_NAME", "TABLE_TYPE"));
getTablesParams.setCurrentUserIdent(ctx.getCurrentUserIdentity().toThrift());
return impl.listTableStatus(getTablesParams);
}
@@ -318,6 +348,17 @@ public class FlightSqlSchemaHelper {
vectorSchemaRoot.setRowCount(getDbsResult.getDbs().size());
}
+ /**
+ * for FlightSqlProducer Schemas.GET_TABLE_TYPES_SCHEMA
+ */
+ public static void getTableTypes(VectorSchemaRoot vectorSchemaRoot) {
+ VarCharVector tableTypeVector = (VarCharVector)
vectorSchemaRoot.getVector("table_type");
+ for (int i = 0; i < TABLE_TYPES.size(); i++) {
+ tableTypeVector.setSafe(i, new Text(TABLE_TYPES.get(i)));
+ }
+ vectorSchemaRoot.setRowCount(TABLE_TYPES.size());
+ }
+
/**
* for FlightSqlProducer Schemas.GET_TABLES_SCHEMA_NO_SCHEMA and
Schemas.GET_TABLES_SCHEMA
*/
@@ -333,12 +374,21 @@ public class FlightSqlSchemaHelper {
for (int dbIndex = 0; dbIndex < getDbsResult.getDbs().size();
dbIndex++) {
String dbName = getDbsResult.getDbs().get(dbIndex);
String catalogName = getDbsResult.isSetCatalogs() ?
getDbsResult.getCatalogs().get(dbIndex) : "";
- TListTableStatusResult listTableStatusResult =
listTableStatus(dbName, catalogName);
+ List<TTableStatus> tables = listTableStatus(dbName,
catalogName).getTables();
+ if (tableTypeFilter != null) {
+ // The service's single-type filter only handles VIEW. Match
all requested types here
+ // against the authorized results, before collecting their
optional schemas.
+ tables = tables.stream().filter(table ->
tableTypeFilter.contains(table.getType()))
+ .collect(Collectors.toList());
+ }
+ if (tables.isEmpty()) {
+ continue;
+ }
Map<String, List<Field>> tableToFields;
if (includeSchema) {
List<String> tablesName = new ArrayList<>();
- for (TTableStatus tableStatus :
listTableStatusResult.getTables()) {
+ for (TTableStatus tableStatus : tables) {
tablesName.add(tableStatus.getName());
}
TDescribeTablesResult describeTablesResult =
describeTables(dbName, catalogName, tablesName);
@@ -347,7 +397,7 @@ public class FlightSqlSchemaHelper {
tableToFields = null;
}
- for (TTableStatus tableStatus : listTableStatusResult.getTables())
{
+ for (TTableStatus tableStatus : tables) {
catalogNameVector.setSafe(tablesCount, new Text(catalogName));
schemaNameVector.setSafe(tablesCount, new Text(dbName));
tableNameVector.setSafe(tablesCount, new
Text(tableStatus.getName()));
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 139a823ce51..9356c237b6f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -756,8 +756,13 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
// the version needs to be updated.
// Otherwise, the version will always be the old value
// unless there is a query for the table in the
follower node.
+ // UPDATE_TIME is the only status column that reads
the partition versions
+ // (OlapTable.getUpdateTime() takes the partitions'
visible version time), so a
+ // request that does not need it, such as Flight SQL
GetTables asking for names
+ // and types, does not wait on the meta service.
if (!Env.getCurrentEnv().isMaster() &&
Config.isCloudMode()
- && table instanceof OlapTable) {
+ && table instanceof OlapTable
+ && needTableStatusColumn(requiredColumns,
"UPDATE_TIME")) {
OlapTable olapTable = (OlapTable) table;
List<CloudPartition> partitions =
olapTable.getAllPartitions().stream()
.filter(p -> p instanceof CloudPartition)
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlSchemaHelperTableTypesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlSchemaHelperTableTypesTest.java
new file mode 100644
index 00000000000..3f9cae0087a
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlSchemaHelperTableTypesTest.java
@@ -0,0 +1,177 @@
+// 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.
+
+package org.apache.doris.arrowflight;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.service.ExecuteEnv;
+import org.apache.doris.service.FrontendServiceImpl;
+import org.apache.doris.thrift.TColumnDef;
+import org.apache.doris.thrift.TColumnDesc;
+import org.apache.doris.thrift.TDescribeTablesParams;
+import org.apache.doris.thrift.TDescribeTablesResult;
+import org.apache.doris.thrift.TGetDbsParams;
+import org.apache.doris.thrift.TGetDbsResult;
+import org.apache.doris.thrift.TGetTablesParams;
+import org.apache.doris.thrift.TListTableStatusResult;
+import org.apache.doris.thrift.TPrimitiveType;
+import org.apache.doris.thrift.TTableStatus;
+
+import org.apache.arrow.flight.sql.FlightSqlProducer.Schemas;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ReadChannel;
+import org.apache.arrow.vector.ipc.message.MessageSerializer;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayInputStream;
+import java.nio.channels.Channels;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class FlightSqlSchemaHelperTableTypesTest {
+ private static Stream<Arguments> tableFilters() {
+ return Stream.of(false, true).flatMap(includeSchema -> Stream.of(
+ Arguments.of(includeSchema, Collections.emptyList(),
Arrays.asList("base", "system", "view")),
+ Arguments.of(includeSchema, Arrays.asList("VIEW", "BASE
TABLE"), Arrays.asList("base", "view")),
+ Arguments.of(includeSchema, Arrays.asList("BASE TABLE",
"VIEW"), Arrays.asList("base", "view")),
+ Arguments.of(includeSchema, Arrays.asList("BASE TABLE"),
Arrays.asList("base")),
+ Arguments.of(includeSchema, Arrays.asList("SYSTEM VIEW"),
Arrays.asList("system")),
+ Arguments.of(includeSchema, Arrays.asList("VIEW"),
Arrays.asList("view")),
+ Arguments.of(includeSchema, Arrays.asList("UNKNOWN"),
Collections.emptyList()),
+ Arguments.of(includeSchema, Arrays.asList("UNKNOWN", "VIEW",
"VIEW"), Arrays.asList("view")),
+ // The names Flight SQL calls common, in the spellings clients
send: the ADBC Flight SQL
+ // driver asks for "table", JDBC tools for "TABLE".
+ Arguments.of(includeSchema, Arrays.asList("table"),
Arrays.asList("base")),
+ Arguments.of(includeSchema, Arrays.asList("TABLE", "VIEW"),
Arrays.asList("base", "view")),
+ Arguments.of(includeSchema, Arrays.asList("view"),
Arrays.asList("view")),
+ Arguments.of(includeSchema, Arrays.asList("SYSTEM TABLE"),
Arrays.asList("system")),
+ Arguments.of(includeSchema, Arrays.asList("base table",
"System View"),
+ Arrays.asList("base", "system"))));
+ }
+
+ @ParameterizedTest
+ @MethodSource("tableFilters")
+ public void filtersAuthorizedTablesBeforeDescribing(boolean includeSchema,
List<String> types,
+ List<String> expectedNames) throws Exception {
+ checkTables("internal", includeSchema, types, expectedNames);
+ }
+
+ @Test
+ public void retainsExternalCatalogForFilteredSchemas() throws Exception {
+ checkTables("external", true, Arrays.asList("VIEW", "BASE TABLE"),
Arrays.asList("base", "view"));
+ }
+
+ private void checkTables(String catalog, boolean includeSchema,
List<String> types,
+ List<String> expectedNames) throws Exception {
+ ConnectContext ctx = Mockito.mock(ConnectContext.class);
+
Mockito.when(ctx.getCurrentUserIdentity()).thenReturn(UserIdentity.ROOT);
+ List<TTableStatus> authorizedTables = Arrays.asList(
+ new TTableStatus().setName("base").setType("BASE TABLE"),
+ new TTableStatus().setName("system").setType("SYSTEM VIEW"),
+ new TTableStatus().setName("view").setType("VIEW"));
+ List<String> describedNames = new ArrayList<>();
+ try (MockedStatic<ExecuteEnv> executeEnv =
Mockito.mockStatic(ExecuteEnv.class);
+ MockedConstruction<FrontendServiceImpl> services =
Mockito.mockConstruction(
+ FrontendServiceImpl.class, (service, construction) -> {
+
Mockito.when(service.getDbNames(Mockito.any(TGetDbsParams.class))).thenAnswer(invocation
-> {
+ TGetDbsParams params =
invocation.getArgument(0);
+ Assertions.assertEquals(catalog,
params.getCatalog());
+ Assertions.assertEquals("db%",
params.getPattern());
+
Assertions.assertEquals(UserIdentity.ROOT.toThrift(),
params.getCurrentUserIdent());
+ return new
TGetDbsResult().setDbs(Arrays.asList("db"))
+ .setCatalogs(Arrays.asList(catalog));
+ });
+
Mockito.when(service.listTableStatus(Mockito.any(TGetTablesParams.class)))
+ .thenAnswer(invocation -> {
+ TGetTablesParams params =
invocation.getArgument(0);
+ Assertions.assertEquals(catalog,
params.getCatalog());
+ Assertions.assertEquals("db",
params.getDb());
+ Assertions.assertEquals("%",
params.getPattern());
+
Assertions.assertEquals(UserIdentity.ROOT.toThrift(),
+ params.getCurrentUserIdent());
+ // Names and types only: no table
statistics, and no UPDATE_TIME, the one
+ // column a cloud follower refreshes
partition versions from the meta
+ // service for.
+
Assertions.assertEquals(Set.of("TABLE_NAME", "TABLE_TYPE"),
+ params.getRequiredColumns());
+ // The service only recognizes VIEW;
other types return all authorized tables.
+ return new
TListTableStatusResult().setTables(authorizedTables.stream()
+ .filter(table ->
!"VIEW".equals(params.getType())
+ ||
"VIEW".equals(table.getType()))
+ .collect(Collectors.toList()));
+ });
+
Mockito.when(service.describeTables(Mockito.any(TDescribeTablesParams.class)))
+ .thenAnswer(invocation -> {
+ TDescribeTablesParams params =
invocation.getArgument(0);
+
describedNames.addAll(params.getTablesName());
+ Assertions.assertEquals(catalog,
params.getCatalog());
+
Assertions.assertEquals(UserIdentity.ROOT.toThrift(),
+ params.getCurrentUserIdent());
+ List<TColumnDef> columns = new
ArrayList<>();
+ List<Integer> offsets = new
ArrayList<>();
+ for (String name :
params.getTablesName()) {
+ columns.add(new TColumnDef(new
TColumnDesc(name + "_id", TPrimitiveType.INT)));
+ offsets.add(columns.size());
+ }
+ return new
TDescribeTablesResult().setColumns(columns).setTablesOffset(offsets);
+ });
+ });
+ RootAllocator allocator = new RootAllocator();
+ VectorSchemaRoot root = VectorSchemaRoot.create(includeSchema
+ ? Schemas.GET_TABLES_SCHEMA :
Schemas.GET_TABLES_SCHEMA_NO_SCHEMA, allocator)) {
+ FlightSqlSchemaHelper helper = new FlightSqlSchemaHelper(ctx);
+
helper.setParameterForGetTables(CommandGetTables.newBuilder().setCatalog(catalog)
+
.setDbSchemaFilterPattern("db%").setTableNameFilterPattern("%")
+
.addAllTableTypes(types).setIncludeSchema(includeSchema).build());
+ root.allocateNew();
+ helper.getTables(root);
+ List<String> actualNames = new ArrayList<>();
+ for (int row = 0; row < root.getRowCount(); row++) {
+
actualNames.add(root.getVector("table_name").getObject(row).toString());
+ if (includeSchema) {
+ byte[] bytes = ((VarBinaryVector)
root.getVector("table_schema")).get(row);
+ Schema schema = MessageSerializer.deserializeSchema(
+ new ReadChannel(Channels.newChannel(new
ByteArrayInputStream(bytes))));
+ Assertions.assertEquals(1, schema.getFields().size());
+ Assertions.assertEquals(actualNames.get(row) + "_id",
schema.getFields().get(0).getName());
+ Assertions.assertEquals(new ArrowType.Int(32, true),
schema.getFields().get(0).getType());
+ }
+ }
+ Assertions.assertEquals(expectedNames, actualNames);
+ Assertions.assertEquals(includeSchema ? expectedNames :
Collections.emptyList(), describedNames);
+
Mockito.verify(services.constructed().get(0)).listTableStatus(Mockito.any(TGetTablesParams.class));
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlTableTypesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlTableTypesTest.java
new file mode 100644
index 00000000000..ec6c4a44e24
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/arrowflight/FlightSqlTableTypesTest.java
@@ -0,0 +1,151 @@
+// 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.
+
+package org.apache.doris.arrowflight;
+
+import org.apache.doris.arrowflight.sessions.FlightSessionsManager;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.qe.ConnectContext;
+
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightProducer.CallContext;
+import org.apache.arrow.flight.FlightProducer.ServerStreamListener;
+import org.apache.arrow.flight.FlightServer;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.Location;
+import org.apache.arrow.flight.sql.FlightSqlClient;
+import org.apache.arrow.flight.sql.FlightSqlProducer.Schemas;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class FlightSqlTableTypesTest {
+ @Test
+ public void streamsSortedDorisTypesAndReleasesBuffers() throws Exception {
+ checkStream(false);
+ }
+
+ @Test
+ public void releasesBuffersWhenSendingFails() throws Exception {
+ checkStream(true);
+ }
+
+ // GetFlightInfo and DoGet through a real Flight SQL server, on repeated
calls: the info carries the
+ // GetTableTypes schema and one endpoint, and the stream returns the same
sorted types every time.
+ @Test
+ public void tableTypesAreSortedAndStableThroughFlightRpc() throws
Exception {
+ boolean previous = FeConstants.runningUnitTest;
+ FeConstants.runningUnitTest = true;
+ ConnectContext ctx = ConnectContext.forFlight("metadata-token");
+ try {
+ FlightSessionsManager sessions =
Mockito.mock(FlightSessionsManager.class);
+
Mockito.when(sessions.getConnectContext(Mockito.anyString())).thenReturn(ctx);
+ Location location = Location.forGrpcInsecure("127.0.0.1", 0);
+ try (DorisFlightSqlProducer producer = new
DorisFlightSqlProducer(location, sessions);
+ RootAllocator allocator = new RootAllocator();
+ FlightServer server = FlightServer.builder(allocator,
location, producer).build().start();
+ FlightClient client = FlightClient.builder(allocator,
+ Location.forGrpcInsecure("127.0.0.1",
server.getPort())).build()) {
+ FlightSqlClient sql = new FlightSqlClient(client);
+ for (int iteration = 0; iteration < 3; ++iteration) {
+ FlightInfo info = sql.getTableTypes();
+ Assertions.assertEquals(Schemas.GET_TABLE_TYPES_SCHEMA,
info.getSchema());
+ Assertions.assertEquals(1, info.getEndpoints().size());
+ List<String> types = new ArrayList<>();
+ try (FlightStream stream =
client.getStream(info.getEndpoints().get(0).getTicket())) {
+ Assertions.assertEquals(info.getSchema(),
stream.getSchema());
+ while (stream.next()) {
+ VectorSchemaRoot root = stream.getRoot();
+ for (int row = 0; row < root.getRowCount(); ++row)
{
+
types.add(root.getVector("table_type").getObject(row).toString());
+ }
+ }
+ }
+ Assertions.assertEquals(Arrays.asList("BASE TABLE",
"SYSTEM VIEW", "VIEW"), types);
+ }
+ // GetFlightInfo answers from the static schema; only each
DoGet resolves the session.
+ Mockito.verify(sessions,
Mockito.times(3)).getConnectContext(Mockito.anyString());
+ }
+ } finally {
+ ctx.getFlightSqlChannel().close();
+ FeConstants.runningUnitTest = previous;
+ }
+ }
+
+ private void checkStream(boolean failSend) throws Exception {
+ boolean previous = FeConstants.runningUnitTest;
+ FeConstants.runningUnitTest = true;
+ try {
+ ConnectContext ctx = ConnectContext.forFlight("metadata-token");
+ FlightSessionsManager sessions =
Mockito.mock(FlightSessionsManager.class);
+
Mockito.when(sessions.getConnectContext("metadata-token")).thenReturn(ctx);
+ CallContext context = Mockito.mock(CallContext.class);
+ Mockito.when(context.peerIdentity()).thenReturn("metadata-token");
+ try (DorisFlightSqlProducer producer = new DorisFlightSqlProducer(
+ Location.forGrpcInsecure("127.0.0.1", 9090), sessions)) {
+ BufferAllocator allocator = Deencapsulation.getField(producer,
"rootAllocator");
+ for (int i = 0; i < 10; i++) {
+ ServerStreamListener listener =
Mockito.mock(ServerStreamListener.class);
+ AtomicReference<VectorSchemaRoot> batch = new
AtomicReference<>();
+ List<String> values = new ArrayList<>();
+ Mockito.doAnswer(invocation -> {
+ batch.set(invocation.getArgument(0));
+
Assertions.assertEquals(Schemas.GET_TABLE_TYPES_SCHEMA,
batch.get().getSchema());
+ return null;
+
}).when(listener).start(Mockito.any(VectorSchemaRoot.class));
+ Mockito.doAnswer(invocation -> {
+ VectorSchemaRoot root = batch.get();
+ for (int row = 0; row < root.getRowCount(); row++) {
+
values.add(root.getVector("table_type").getObject(row).toString());
+ }
+ if (failSend) {
+ throw new IllegalStateException("simulated send
failure");
+ }
+ return null;
+ }).when(listener).putNext();
+ if (failSend) {
+ Assertions.assertThrows(RuntimeException.class,
+ () -> producer.getStreamTableTypes(context,
listener));
+
Mockito.verify(listener).error(Mockito.any(Throwable.class));
+ Mockito.verify(listener, Mockito.never()).completed();
+ } else {
+ producer.getStreamTableTypes(context, listener);
+ Mockito.verify(listener).completed();
+ Mockito.verify(listener,
Mockito.never()).error(Mockito.any(Throwable.class));
+ }
+ Assertions.assertEquals(Arrays.asList("BASE TABLE",
"SYSTEM VIEW", "VIEW"), values);
+ Assertions.assertEquals(0, allocator.getAllocatedMemory());
+ }
+ Mockito.verify(sessions,
Mockito.times(10)).getConnectContext("metadata-token");
+ } finally {
+ ctx.getFlightSqlChannel().close();
+ }
+ } finally {
+ FeConstants.runningUnitTest = previous;
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
index 6c139546424..a55eded6c6a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java
@@ -25,6 +25,7 @@ import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.catalog.Tablet;
+import org.apache.doris.cloud.catalog.CloudPartition;
import org.apache.doris.common.AuthenticationException;
import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
@@ -37,6 +38,7 @@ import org.apache.doris.nereids.trees.plans.commands.Command;
import org.apache.doris.nereids.trees.plans.commands.CreateDatabaseCommand;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.rpc.RpcException;
import org.apache.doris.tablefunction.BackendsTableValuedFunction;
import org.apache.doris.thrift.TBackendsMetadataParams;
import org.apache.doris.thrift.TCheckAuthRequest;
@@ -78,6 +80,7 @@ import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.WriteBlockAllocatingTransaction;
import org.apache.doris.utframe.TestWithFeService;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import org.apache.logging.log4j.Level;
import org.apache.thrift.TException;
@@ -521,6 +524,60 @@ public class FrontendServiceImplTest extends
TestWithFeService {
}
}
+ /**
+ * On a follower in cloud mode, listing a table refreshes its partition
versions from the meta service, which
+ * only UPDATE_TIME reads. A listing that asks for names and types, as
Flight SQL GetTables does, must not
+ * depend on that RPC: when it fails, the enumeration stops and the caller
gets only the rows before it.
+ */
+ @Test
+ public void
testListTableStatusRefreshesCloudFollowerVersionsOnlyForUpdateTime() throws
Exception {
+ createTable("CREATE TABLE test.follower_status_table(\n"
+ + " k1 INT,\n"
+ + " v1 INT\n"
+ + ")\n"
+ + "DUPLICATE KEY(k1)\n"
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ + "PROPERTIES(\"replication_num\" = \"1\");");
+ createView("CREATE VIEW test.follower_status_view AS SELECT k1 FROM
test.follower_status_table");
+ TGetTablesParams namesAndTypes = followerStatusParams("TABLE_NAME",
"TABLE_TYPE");
+ TGetTablesParams updateTimes = followerStatusParams("UPDATE_TIME");
+ Env followerEnv = Mockito.spy(Env.getCurrentEnv());
+ Mockito.doReturn(false).when(followerEnv).isMaster();
+ FrontendServiceImpl impl = new FrontendServiceImpl(exeEnv);
+ try (MockedStatic<Env> env = Mockito.mockStatic(Env.class,
Mockito.CALLS_REAL_METHODS);
+ MockedStatic<Config> config = Mockito.mockStatic(Config.class,
Mockito.CALLS_REAL_METHODS);
+ MockedStatic<CloudPartition> versions =
Mockito.mockStatic(CloudPartition.class)) {
+ env.when(Env::getCurrentEnv).thenReturn(followerEnv);
+ config.when(Config::isCloudMode).thenReturn(true);
+ // The meta service is down: every partition version refresh fails.
+ versions.when(() ->
CloudPartition.getSnapshotVisibleVersion(Mockito.anyList()))
+ .thenThrow(new RpcException("", "injected meta service
failure"));
+
+ TListTableStatusResult listed =
impl.listTableStatus(namesAndTypes);
+
+ Assertions.assertEquals(
+ ImmutableMap.of("follower_status_table", "BASE TABLE",
"follower_status_view", "VIEW"),
+ listed.getTables().stream()
+ .collect(Collectors.toMap(TTableStatus::getName,
TTableStatus::getType)));
+ versions.verify(() ->
CloudPartition.getSnapshotVisibleVersion(Mockito.anyList()), Mockito.never());
+
+ // UPDATE_TIME reads the partitions' visible version time, so it
still refreshes them first.
+ impl.listTableStatus(updateTimes);
+
+ versions.verify(() ->
CloudPartition.getSnapshotVisibleVersion(Mockito.anyList()));
+ }
+ }
+
+ private TGetTablesParams followerStatusParams(String... requiredColumns) {
+ TGetTablesParams params = new TGetTablesParams();
+ params.setCatalog(InternalCatalog.INTERNAL_CATALOG_NAME);
+ params.setDb("test");
+ params.setPattern("follower_status_%");
+ params.setRequiredColumns(Sets.newHashSet(requiredColumns));
+
params.setCurrentUserIdent(connectContext.getCurrentUserIdentity().toThrift());
+ return params;
+ }
+
@Test
public void testCreatePartitionRange() throws Exception {
diff --git
a/regression-test/suites/arrow_flight_sql_p0/test_table_types_over_arrow_flight.groovy
b/regression-test/suites/arrow_flight_sql_p0/test_table_types_over_arrow_flight.groovy
new file mode 100644
index 00000000000..ec5f53f4c50
--- /dev/null
+++
b/regression-test/suites/arrow_flight_sql_p0/test_table_types_over_arrow_flight.groovy
@@ -0,0 +1,141 @@
+// 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.org.apache.arrow.flight.CallOption
+import org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CallOptions
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.flight.CloseSessionRequest
+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.memory.RootAllocator
+import
org.apache.arrow.driver.jdbc.shaded.org.apache.arrow.vector.types.pojo.ArrowType
+import org.apache.doris.regression.util.JdbcUtils
+
+import java.util.concurrent.TimeUnit
+
+suite("test_table_types_over_arrow_flight", "arrow_flight_sql") {
+ def flightConn = context.getArrowFlightSqlConnection()
+ def runOnFlight = { String statement ->
+ JdbcUtils.executeToList(flightConn, statement)
+ }
+ runOnFlight "USE `${context.dbName}`"
+ runOnFlight "DROP VIEW IF EXISTS test_flight_table_types_view"
+ runOnFlight "DROP TABLE IF EXISTS test_flight_table_types_base"
+ runOnFlight """
+ CREATE TABLE test_flight_table_types_base (id INT)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ """
+ runOnFlight """
+ CREATE VIEW test_flight_table_types_view AS
+ SELECT id FROM test_flight_table_types_base
+ """
+
+ // These JDBC metadata methods consume Flight GetTableTypes/GetTables
streams.
+ // SQL result snapshots alone do not exercise those metadata RPCs.
+ def metadata = flightConn.getMetaData()
+ def typeRows = metadata.getTableTypes()
+ try {
+ def types = []
+ while (typeRows.next()) {
+ types.add(typeRows.getString("TABLE_TYPE"))
+ }
+ assertEquals(["BASE TABLE", "SYSTEM VIEW", "VIEW"], types)
+ } finally {
+ typeRows.close()
+ }
+
+ def getTables = { List<String> types ->
+ def rows = metadata.getTables("internal", context.dbName,
"test_flight_table_types%",
+ types == null ? null : types.toArray(new String[0]))
+ try {
+ def tables = []
+ while (rows.next()) {
+ assertEquals("internal", rows.getString("TABLE_CAT"))
+ assertEquals(context.dbName, rows.getString("TABLE_SCHEM"))
+ tables.add([rows.getString("TABLE_NAME"),
rows.getString("TABLE_TYPE")])
+ }
+ return tables.sort { left, right -> left[0] <=> right[0] }
+ } finally {
+ rows.close()
+ }
+ }
+
+ def baseTable = ["test_flight_table_types_base", "BASE TABLE"]
+ def view = ["test_flight_table_types_view", "VIEW"]
+ assertEquals([baseTable, view], getTables(null))
+ assertEquals([baseTable, view], getTables([]))
+ assertEquals([baseTable], getTables(["BASE TABLE"]))
+ assertEquals([view], getTables(["VIEW"]))
+ // Both orders must work: forwarding only the first type loses or leaks
rows.
+ assertEquals([baseTable, view], getTables(["VIEW", "BASE TABLE"]))
+ assertEquals([baseTable, view], getTables(["BASE TABLE", "VIEW"]))
+ assertEquals([], getTables(["SYSTEM VIEW"]))
+ assertEquals([], getTables(["UNKNOWN"]))
+ assertEquals([view], getTables(["UNKNOWN", "VIEW", "VIEW"]))
+ // Flight SQL calls TABLE, VIEW and SYSTEM TABLE the commonly supported
names, and clients send them
+ // in either case: the ADBC Flight SQL driver asks for "table", JDBC tools
for "TABLE".
+ assertEquals([baseTable], getTables(["TABLE"]))
+ assertEquals([baseTable], getTables(["table"]))
+ assertEquals([baseTable, view], getTables(["table", "view"]))
+
+ // The same RPCs from a plain Flight SQL client, called repeatedly:
GetFlightInfo carries the
+ // GetTableTypes schema and one endpoint, DoGet returns the same sorted
types every time, and
+ // the session still answers a query afterwards.
+ def config = context.config.otherConfigs
+ def location =
Location.forGrpcInsecure(config.get("extArrowFlightSqlHost"),
+ Integer.parseInt(config.get("extArrowFlightSqlPort")))
+ new RootAllocator().withCloseable { allocator ->
+ FlightClient.builder(allocator, location).build().withCloseable {
flight ->
+ def token =
flight.authenticateBasicToken(config.get("extArrowFlightSqlUser"),
+ config.get("extArrowFlightSqlPassword")).get()
+ CallOption[] options = [token, CallOptions.timeout(30,
TimeUnit.SECONDS)]
+ def client = new FlightSqlClient(flight)
+ try {
+ 3.times {
+ def info = client.getTableTypes(options)
+ assertEquals(["table_type"],
info.getSchema().getFields()*.getName())
+ assertEquals(new ArrowType.Utf8(),
info.getSchema().getFields()[0].getType())
+ assertFalse(info.getSchema().getFields()[0].isNullable())
+ assertEquals(1, info.getEndpoints().size())
+ def types = []
+ flight.getStream(info.getEndpoints()[0].getTicket(),
options).withCloseable { stream ->
+ assertEquals(info.getSchema(), stream.getSchema())
+ while (stream.next()) {
+ def root = stream.getRoot()
+ for (int row = 0; row < root.getRowCount(); ++row)
{
+
types.add(root.getVector("table_type").getObject(row).toString())
+ }
+ }
+ }
+ assertEquals(["BASE TABLE", "SYSTEM VIEW", "VIEW"], types)
+ }
+ def info = client.execute("SHOW VARIABLES LIKE
'query_timeout'", options)
+ int rows = 0
+ flight.getStream(info.getEndpoints()[0].getTicket(),
options).withCloseable { stream ->
+ while (stream.next()) {
+ rows += stream.getRoot().getRowCount()
+ }
+ }
+ assertEquals(1, rows)
+ } finally {
+ // End the session the way the drivers do rather than leaving
it to wait_timeout.
+ flight.closeSession(new CloseSessionRequest(), token)
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
