This is an automated email from the ASF dual-hosted git repository.
diqiu50 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 60b58ef659 [Cherry-pick to branch-1.3] [#12518] fix(trino-connector):
Keep the engine types when applying projections (#12931)
60b58ef659 is described below
commit 60b58ef6596aab448561163b3014677002cf87c7
Author: Yuhui <[email protected]>
AuthorDate: Fri Sep 4 20:55:40 2026 +0800
[Cherry-pick to branch-1.3] [#12518] fix(trino-connector): Keep the engine
types when applying projections (#12931)
**Cherry-pick Information:**
- Original commit: a4bdfda6385264e611a53aaca3fb6b101036b4e3
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: qbhan <[email protected]>
Co-authored-by: Claude Sonnet 5 <[email protected]>
---
.../jdbc-mysql/00015_projection_pushdown.sql | 17 ++
.../jdbc-mysql/00015_projection_pushdown.txt | 12 ++
.../trino/connector/GravitinoMetadata.java | 113 ++++++++---
.../TestGravitinoMetadataApplyProjection.java | 212 +++++++++++++++++++++
4 files changed, 325 insertions(+), 29 deletions(-)
diff --git
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_projection_pushdown.sql
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_projection_pushdown.sql
new file mode 100644
index 0000000000..e1c0c410f2
--- /dev/null
+++
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_projection_pushdown.sql
@@ -0,0 +1,17 @@
+CREATE SCHEMA gt_mysql.gt_db_projection;
+
+CREATE TABLE gt_mysql.gt_db_projection.tb01 (
+ id int,
+ name varchar
+);
+
+insert into gt_mysql.gt_db_projection.tb01(id, name) values (1, 'sam'), (2,
'jerry');
+
+-- Selecting every column together with a computed one pushes the projection
into the internal
+-- connector, which types the unbounded varchar column as varchar(65535).
Since Trino 444 the plan
+-- is rejected if that type reaches the engine instead of the one this
connector declared.
+select *, if(name is null, '', name) from gt_mysql.gt_db_projection.tb01 order
by id;
+
+drop table gt_mysql.gt_db_projection.tb01;
+
+drop schema gt_mysql.gt_db_projection;
diff --git
a/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_projection_pushdown.txt
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_projection_pushdown.txt
new file mode 100644
index 0000000000..741e186fff
--- /dev/null
+++
b/trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets/jdbc-mysql/00015_projection_pushdown.txt
@@ -0,0 +1,12 @@
+CREATE SCHEMA
+
+CREATE TABLE
+
+INSERT: 2 rows
+
+"1","sam","sam"
+"2","jerry","jerry"
+
+DROP TABLE
+
+DROP SCHEMA
diff --git
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata.java
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata.java
index f05b557233..7e74259be8 100644
---
a/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata.java
+++
b/trino-connector/trino-connector/src/main/java/org/apache/gravitino/trino/connector/GravitinoMetadata.java
@@ -59,15 +59,19 @@ import io.trino.spi.connector.SystemTable;
import io.trino.spi.connector.TopNApplicationResult;
import io.trino.spi.expression.ConnectorExpression;
import io.trino.spi.expression.Constant;
+import io.trino.spi.expression.Variable;
import io.trino.spi.function.LanguageFunction;
import io.trino.spi.function.SchemaFunctionName;
import io.trino.spi.security.TrinoPrincipal;
import io.trino.spi.statistics.ColumnStatistics;
import io.trino.spi.statistics.TableStatistics;
import io.trino.spi.type.Type;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Deque;
+import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -502,37 +506,47 @@ public abstract class GravitinoMetadata implements
ConnectorMetadata {
ConnectorTableHandle handle,
List<ConnectorExpression> projections,
Map<String, ColumnHandle> assignments) {
+ Map<String, ColumnHandle> internalAssignments =
+ assignments.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey, entry ->
GravitinoHandle.unWrap(entry.getValue())));
+ // The engine variable name for each column handle passed in, the reverse
of
+ // internalAssignments. Used below to find the engine variable a returned
assignment refers
+ // to even if the internal connector renamed it (e.g. to avoid a name
collision), since the
+ // same underlying column handle is preserved either way.
+ Map<ColumnHandle, String> engineVariableByColumn =
+ internalAssignments.entrySet().stream()
+ .collect(
+ Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey,
(first, second) -> first));
+ SchemaTableName tableName = getTableName(handle);
return internalMetadata
- .applyProjection(
- session,
- GravitinoHandle.unWrap(handle),
- projections,
- assignments.entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey, entry ->
GravitinoHandle.unWrap(entry.getValue()))))
+ .applyProjection(session, GravitinoHandle.unWrap(handle), projections,
internalAssignments)
.map(
- result ->
- new ProjectionApplicationResult<>(
- new GravitinoTableHandle(
- getTableName(handle).getSchemaName(),
- getTableName(handle).getTableName(),
- result.getHandle()),
- result.getProjections(),
- result.getAssignments().stream()
- .map(
- entry ->
- new Assignment(
- entry.getVariable(),
- new GravitinoColumnHandle(
- getColumnName(
- session,
- GravitinoHandle.unWrap(handle),
- entry.getColumn()),
- entry.getColumn()),
- entry.getType()))
- .toList(),
- result.isPrecalculateStatistics()));
+ result -> {
+ // Restore the types the engine assigned to the projected
variables; see
+ // resolveAssignmentType for why this is needed.
+ Map<String, Type> engineTypes =
collectVariableTypes(projections);
+ return new ProjectionApplicationResult<>(
+ new GravitinoTableHandle(
+ tableName.getSchemaName(), tableName.getTableName(),
result.getHandle()),
+ result.getProjections(),
+ result.getAssignments().stream()
+ .map(
+ entry ->
+ new Assignment(
+ entry.getVariable(),
+ new GravitinoColumnHandle(
+ getColumnName(
+ session,
+ GravitinoHandle.unWrap(handle),
+ entry.getColumn()),
+ entry.getColumn()),
+ resolveAssignmentType(
+ entry, engineTypes,
engineVariableByColumn)))
+ .toList(),
+ result.isPrecalculateStatistics());
+ });
}
@Override
@@ -909,4 +923,45 @@ public abstract class GravitinoMetadata implements
ConnectorMetadata {
sb.append(")");
return sb.toString();
}
+
+ /**
+ * Resolves the type for an assignment returned by the internal connector.
The internal connector
+ * may type an assignment from its own column handle, which can disagree
with the type this
+ * connector declared for the same column, for example an unbounded varchar
here and a
+ * varchar(65535) there for MySQL tinytext. Since Trino 444, a plan whose
symbol and expression
+ * types differ is rejected, so the engine type is applied instead - looked
up via the column
+ * handle rather than the assignment's variable name, since the internal
connector may have
+ * renamed the variable (e.g. to avoid a name collision) while keeping the
same underlying column.
+ * Columns the internal connector synthesized, which have no entry in {@code
+ * engineVariableByColumn}, keep their internal types.
+ *
+ * @param assignment the assignment returned by the internal connector
+ * @param engineTypes the types the engine assigned to the projected
variables, by variable name
+ * @param engineVariableByColumn the engine variable name for each column
handle the engine passed
+ * in
+ * @return the type the returned assignment must carry
+ */
+ private static Type resolveAssignmentType(
+ Assignment assignment,
+ Map<String, Type> engineTypes,
+ Map<ColumnHandle, String> engineVariableByColumn) {
+ String engineVariable = engineVariableByColumn.get(assignment.getColumn());
+ if (engineVariable == null) {
+ return assignment.getType();
+ }
+ return engineTypes.getOrDefault(engineVariable, assignment.getType());
+ }
+
+ private static Map<String, Type>
collectVariableTypes(List<ConnectorExpression> expressions) {
+ Map<String, Type> types = new HashMap<>();
+ Deque<ConnectorExpression> pending = new ArrayDeque<>(expressions);
+ while (!pending.isEmpty()) {
+ ConnectorExpression expression = pending.pop();
+ if (expression instanceof Variable) {
+ types.put(((Variable) expression).getName(), expression.getType());
+ }
+ pending.addAll(expression.getChildren());
+ }
+ return types;
+ }
}
diff --git
a/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoMetadataApplyProjection.java
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoMetadataApplyProjection.java
new file mode 100644
index 0000000000..b4c46b5ae5
--- /dev/null
+++
b/trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/TestGravitinoMetadataApplyProjection.java
@@ -0,0 +1,212 @@
+/*
+ * 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.gravitino.trino.connector;
+
+import static io.trino.spi.type.VarcharType.createUnboundedVarcharType;
+import static io.trino.spi.type.VarcharType.createVarcharType;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import io.trino.spi.connector.Assignment;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorMetadata;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableHandle;
+import io.trino.spi.connector.ProjectionApplicationResult;
+import io.trino.spi.expression.Call;
+import io.trino.spi.expression.ConnectorExpression;
+import io.trino.spi.expression.FunctionName;
+import io.trino.spi.expression.Variable;
+import io.trino.spi.type.Type;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadata;
+import
org.apache.gravitino.trino.connector.catalog.CatalogConnectorMetadataAdapter;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that {@link GravitinoMetadata#applyProjection} hands back the types
the engine assigned to
+ * the projected columns, rather than the types of the internal connector's
column handles.
+ */
+public class TestGravitinoMetadataApplyProjection {
+
+ private static final Type UNBOUNDED_VARCHAR = createUnboundedVarcharType();
+ private static final Type BOUNDED_VARCHAR = createVarcharType(255);
+
+ @Test
+ public void testEngineTypeIsRestoredForProjectedColumn() {
+ ColumnHandle internalColumnHandle = mock(ColumnHandle.class);
+ Fixture fixture = new Fixture();
+ fixture.declareColumn("col_tinytext", internalColumnHandle);
+ fixture.internalReturns(
+ List.of(new Variable("col_tinytext", UNBOUNDED_VARCHAR)),
+ List.of(new Assignment("col_tinytext", internalColumnHandle,
BOUNDED_VARCHAR)));
+
+ List<Assignment> assignments =
+ fixture.applyProjection(List.of(new Variable("col_tinytext",
UNBOUNDED_VARCHAR)));
+
+ assertEquals(1, assignments.size());
+ assertEquals(UNBOUNDED_VARCHAR, assignments.get(0).getType());
+ assertEquals("col_tinytext", assignments.get(0).getVariable());
+ }
+
+ @Test
+ public void testEngineTypeIsRestoredForRenamedColumn() {
+ ColumnHandle internalColumnHandle = mock(ColumnHandle.class);
+ Fixture fixture = new Fixture();
+ fixture.declareColumn("col_tinytext", internalColumnHandle);
+ // The internal connector renames the variable (e.g. to avoid a name
collision) but keeps the
+ // same underlying column handle, matching the "col_tinytext_7" symbol
from issue #12518.
+ fixture.internalReturns(
+ List.of(new Variable("col_tinytext_7", BOUNDED_VARCHAR)),
+ List.of(new Assignment("col_tinytext_7", internalColumnHandle,
BOUNDED_VARCHAR)));
+
+ List<Assignment> assignments =
+ fixture.applyProjection(List.of(new Variable("col_tinytext",
UNBOUNDED_VARCHAR)));
+
+ assertEquals(1, assignments.size());
+ assertEquals(UNBOUNDED_VARCHAR, assignments.get(0).getType());
+ assertEquals("col_tinytext_7", assignments.get(0).getVariable());
+ }
+
+ @Test
+ public void testEngineTypeIsRestoredForColumnNestedInExpression() {
+ ColumnHandle internalColumnHandle = mock(ColumnHandle.class);
+ Fixture fixture = new Fixture();
+ fixture.declareColumn("col_text", internalColumnHandle);
+ fixture.internalReturns(
+ List.of(new Variable("col_text", UNBOUNDED_VARCHAR)),
+ List.of(new Assignment("col_text", internalColumnHandle,
createVarcharType(65535))));
+
+ // The column only appears as an argument of a function call, never on its
own.
+ ConnectorExpression call =
+ new Call(
+ UNBOUNDED_VARCHAR,
+ new FunctionName("lower"),
+ List.of(new Variable("col_text", UNBOUNDED_VARCHAR)));
+
+ List<Assignment> assignments = fixture.applyProjection(List.of(call));
+
+ assertEquals(1, assignments.size());
+ assertEquals(UNBOUNDED_VARCHAR, assignments.get(0).getType());
+ }
+
+ @Test
+ public void testSyntheticColumnKeepsInternalType() {
+ ColumnHandle internalColumnHandle = mock(ColumnHandle.class);
+ Fixture fixture = new Fixture();
+ // A column synthesized by the internal connector has no counterpart among
the engine variables.
+ fixture.declareColumn("expr_1", internalColumnHandle);
+ fixture.internalReturns(
+ List.of(new Variable("expr_1", BOUNDED_VARCHAR)),
+ List.of(new Assignment("expr_1", internalColumnHandle,
BOUNDED_VARCHAR)));
+
+ List<Assignment> assignments =
+ fixture.applyProjection(List.of(new Variable("col_tinytext",
UNBOUNDED_VARCHAR)));
+
+ assertEquals(1, assignments.size());
+ assertEquals(BOUNDED_VARCHAR, assignments.get(0).getType());
+ }
+
+ @Test
+ public void testSynthesizedColumnReusingAnEngineNameKeepsInternalType() {
+ ColumnHandle inputHandle = mock(ColumnHandle.class);
+ ColumnHandle synthesizedHandle = mock(ColumnHandle.class);
+ Fixture fixture = new Fixture();
+ fixture.declareColumn("expr_1", inputHandle);
+ fixture.declareColumn("expr_1_synthesized", synthesizedHandle);
+ // The internal connector reuses the engine variable name for a column it
synthesized itself.
+ fixture.internalReturns(
+ List.of(new Variable("expr_1", BOUNDED_VARCHAR)),
+ List.of(new Assignment("expr_1", synthesizedHandle, BOUNDED_VARCHAR)));
+
+ List<Assignment> assignments =
+ fixture.applyProjection(List.of(new Variable("expr_1",
UNBOUNDED_VARCHAR)));
+
+ assertEquals(1, assignments.size());
+ assertEquals(BOUNDED_VARCHAR, assignments.get(0).getType());
+ }
+
+ @Test
+ public void testEmptyResultIsPassedThrough() {
+ Fixture fixture = new Fixture();
+ when(fixture.internalMetadata.applyProjection(
+ any(ConnectorSession.class), any(ConnectorTableHandle.class),
anyList(), anyMap()))
+ .thenReturn(Optional.empty());
+
+ assertFalse(
+ fixture
+ .metadata
+ .applyProjection(
+ fixture.session,
+ fixture.tableHandle,
+ List.of(new Variable("col_tinytext", UNBOUNDED_VARCHAR)),
+ Map.of())
+ .isPresent());
+ }
+
+ private static final class Fixture {
+ private final ConnectorMetadata internalMetadata =
mock(ConnectorMetadata.class);
+ private final ConnectorSession session = mock(ConnectorSession.class);
+ private final ConnectorTableHandle internalTableHandle =
mock(ConnectorTableHandle.class);
+ private final GravitinoTableHandle tableHandle;
+ private final GravitinoMetadata metadata;
+ private final Map<String, ColumnHandle> assignments = new HashMap<>();
+
+ private Fixture() {
+ tableHandle = new GravitinoTableHandle("test_schema", "test_table",
internalTableHandle);
+ metadata =
+ new GravitinoMetadata(
+ mock(CatalogConnectorMetadata.class),
+ mock(CatalogConnectorMetadataAdapter.class),
+ internalMetadata) {};
+ }
+
+ private void declareColumn(String columnName, ColumnHandle
internalColumnHandle) {
+ when(internalMetadata.getColumnMetadata(session, internalTableHandle,
internalColumnHandle))
+ .thenReturn(new ColumnMetadata(columnName, UNBOUNDED_VARCHAR));
+ assignments.put(columnName, new GravitinoColumnHandle(columnName,
internalColumnHandle));
+ }
+
+ private void internalReturns(List<ConnectorExpression> projections,
List<Assignment> results) {
+ when(internalMetadata.applyProjection(
+ any(ConnectorSession.class), any(ConnectorTableHandle.class),
anyList(), anyMap()))
+ .thenReturn(
+ Optional.of(
+ new ProjectionApplicationResult<>(
+ internalTableHandle, projections, results, false)));
+ }
+
+ private List<Assignment> applyProjection(List<ConnectorExpression>
projections) {
+ Optional<ProjectionApplicationResult<ConnectorTableHandle>> result =
+ metadata.applyProjection(session, tableHandle, projections,
assignments);
+ assertTrue(result.isPresent());
+ return result.get().getAssignments();
+ }
+ }
+}
