Skip to content

fixed for trino 476 when use jdk24 #106

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
awol2005ex wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fixed for trino 476 when use jdk24 #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
14268cb
fixed for trino 476 when use jdk24
awol2005ex Jun 12, 2025
953132d
remainExpression null bug
awol2005ex Jul 9, 2025
194e780
Merge branch 'main' into awol2005ex-br-trino476
awol2005ex Jul 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 3 additions & 8 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,11 @@ under the License.
<paimon.version>1.0.0</paimon.version>
<target.java.version>21</target.java.version>
<jdk.test.version>21</jdk.test.version>
<trino.version>440</trino.version>
<airlift.version>241</airlift.version>
<trino.version>476</trino.version>
<airlift.version>336</airlift.version>
<slice.version>2.2</slice.version>
<maven.toolchains.plugin.version>3.1.0</maven.toolchains.plugin.version>
<hadoop.apache.version>3.2.0-18</hadoop.apache.version>
<hadoop.apache.version>3.3.5-3</hadoop.apache.version>
<junit5.version>5.8.1</junit5.version>
<slf4j.version>2.0.13</slf4j.version>
<guava.version>31.1-jre</guava.version>
Expand Down Expand Up @@ -97,11 +97,6 @@ under the License.
</exclusions>
</dependency>

<dependency>
<groupId>io.trino</groupId>
<artifactId>trino-hdfs</artifactId>
<version>${trino.version}</version>
</dependency>

<dependency>
<groupId>io.trino</groupId>
Expand Down
9 changes: 5 additions & 4 deletions src/main/java/org/apache/paimon/trino/DirectTrinoPageSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@

package org.apache.paimon.trino;

import io.trino.spi.Page;
import io.trino.spi.connector.ConnectorPageSource;
import io.trino.spi.connector.SourcePage;
import io.trino.spi.metrics.Metrics;

import java.io.IOException;
Expand Down Expand Up @@ -54,15 +54,16 @@ public boolean isFinished() {
}

@Override
public Page getNextPage() {
public SourcePage getNextSourcePage()
{
try {
if (current == null) {
return null;
}
Page dataPage = current.getNextPage();
SourcePage dataPage = current.getNextSourcePage();
if (dataPage == null) {
advance();
return getNextPage();
return getNextSourcePage();
}

return dataPage;
Expand Down
19 changes: 12 additions & 7 deletions src/main/java/org/apache/paimon/trino/TrinoConnectorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,6 @@

package org.apache.paimon.trino;

import org.apache.paimon.utils.StringUtils;

import com.google.inject.Binder;
import com.google.inject.Injector;
import com.google.inject.Key;
Expand All @@ -29,8 +27,6 @@
import io.opentelemetry.api.OpenTelemetry;
import io.opentelemetry.api.trace.Tracer;
import io.trino.filesystem.manager.FileSystemModule;
import io.trino.hdfs.HdfsModule;
import io.trino.hdfs.authentication.HdfsAuthenticationModule;
import io.trino.plugin.base.classloader.ClassLoaderSafeConnectorMetadata;
import io.trino.plugin.base.classloader.ClassLoaderSafeConnectorPageSinkProvider;
import io.trino.plugin.base.classloader.ClassLoaderSafeConnectorPageSourceProvider;
Expand All @@ -44,6 +40,7 @@
import io.trino.spi.function.FunctionProvider;
import io.trino.spi.function.table.ConnectorTableFunction;
import io.trino.spi.type.TypeManager;
import org.apache.paimon.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.w3c.dom.Element;
Expand Down Expand Up @@ -104,8 +101,6 @@ public Connector create(
new Bootstrap(
new JsonModule(),
new TrinoModule(config),
new HdfsModule(),
new HdfsAuthenticationModule(),
// bind the trino file system module
newFileSystemModule(catalogName, context),
binder -> {
Expand Down Expand Up @@ -198,12 +193,22 @@ private static FileSystemModule newFileSystemModule(
try {
if (constructor.getParameterCount() == 0) {
return (FileSystemModule) constructor.newInstance();
} else {
}
else if (constructor.getParameterCount() == 3) {
// for trino 440
return (FileSystemModule)
constructor.newInstance(
catalogName, context.getNodeManager(), context.getOpenTelemetry());
}
else if (constructor.getParameterCount() == 4) {
// for trino 476
return (FileSystemModule)
constructor.newInstance(
catalogName, context.getNodeManager(), context.getOpenTelemetry(), false);
}
else {
throw new RuntimeException("Unsupported trino version");
}
} catch (InstantiationException | InvocationTargetException | IllegalAccessException e) {
throw new RuntimeException(e);
}
Expand Down
60 changes: 34 additions & 26 deletions src/main/java/org/apache/paimon/trino/TrinoMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,30 +18,12 @@

package org.apache.paimon.trino;

import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.fs.Path;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.paimon.trino.catalog.TrinoCatalog;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.InstantiationUtil;
import org.apache.paimon.utils.StringUtils;

import io.airlift.slice.Slice;
import io.trino.spi.TrinoException;
import io.trino.spi.connector.Assignment;
import io.trino.spi.connector.ColumnHandle;
import io.trino.spi.connector.ColumnMetadata;
import io.trino.spi.connector.ColumnPosition;
import io.trino.spi.connector.ConnectorInsertTableHandle;
import io.trino.spi.connector.ConnectorMergeTableHandle;
import io.trino.spi.connector.ConnectorMetadata;
Expand All @@ -60,6 +42,7 @@
import io.trino.spi.connector.ProjectionApplicationResult;
import io.trino.spi.connector.RetryMode;
import io.trino.spi.connector.RowChangeParadigm;
import io.trino.spi.connector.SaveMode;
import io.trino.spi.connector.SchemaTableName;
import io.trino.spi.connector.SchemaTablePrefix;
import io.trino.spi.expression.ConnectorExpression;
Expand All @@ -70,6 +53,24 @@
import io.trino.spi.type.TimestampWithTimeZoneType;
import io.trino.spi.type.Type;
import io.trino.spi.type.VarcharType;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.fs.Path;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageSerializer;
import org.apache.paimon.trino.catalog.TrinoCatalog;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.InstantiationUtil;
import org.apache.paimon.utils.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
Expand Down Expand Up @@ -145,8 +146,9 @@ public ConnectorOutputTableHandle beginCreateTable(
ConnectorSession session,
ConnectorTableMetadata tableMetadata,
Optional<ConnectorTableLayout> layout,
RetryMode retryMode) {
createTable(session, tableMetadata, false);
RetryMode retryMode, boolean replace)
{
createTable(session, tableMetadata, replace ? SaveMode.REPLACE : SaveMode.IGNORE);
return getTableHandle(session, tableMetadata.getTable(), Collections.emptyMap());
}

Expand Down Expand Up @@ -175,6 +177,7 @@ public ConnectorInsertTableHandle beginInsert(
public Optional<ConnectorOutputMetadata> finishInsert(
ConnectorSession session,
ConnectorInsertTableHandle insertHandle,
List<ConnectorTableHandle> sourceTableHandles,
Collection<Slice> fragments,
Collection<ComputedStatistics> computedStatistics) {
return commit(session, (TrinoTableHandle) insertHandle, fragments);
Expand Down Expand Up @@ -263,14 +266,16 @@ public Optional<ConnectorPartitioningHandle> getUpdateLayout(

@Override
public ConnectorMergeTableHandle beginMerge(
ConnectorSession session, ConnectorTableHandle tableHandle, RetryMode retryMode) {
ConnectorSession session, ConnectorTableHandle tableHandle, Map<Integer, Collection<ColumnHandle>> updateCaseColumns, RetryMode retryMode)
{
return new TrinoMergeTableHandle((TrinoTableHandle) tableHandle);
}

@Override
public void finishMerge(
ConnectorSession session,
ConnectorMergeTableHandle mergeTableHandle,
List<ConnectorTableHandle> sourceTableHandles,
Collection<Slice> fragments,
Collection<ComputedStatistics> computedStatistics) {
commit(session, (TrinoTableHandle) mergeTableHandle.getTableHandle(), fragments);
Expand Down Expand Up @@ -413,7 +418,7 @@ public ConnectorTableHandle getTableHandle(
return getTableHandle(session, tableName, dynamicOptions);
}

@Override

public TrinoTableHandle getTableHandle(ConnectorSession session, SchemaTableName tableName) {
return getTableHandle(session, tableName, Collections.emptyMap());
}
Expand Down Expand Up @@ -494,13 +499,14 @@ private List<SchemaTableName> listTables(String schema) {
public void createTable(
ConnectorSession session,
ConnectorTableMetadata tableMetadata,
boolean ignoreExisting) {
SaveMode saveMode)
{
SchemaTableName table = tableMetadata.getTable();
Identifier identifier = Identifier.create(table.getSchemaName(), table.getTableName());

try {
catalog.initSession(session);
catalog.createTable(identifier, prepareSchema(tableMetadata), false);
catalog.createTable(identifier, prepareSchema(tableMetadata), saveMode.equals(SaveMode.IGNORE));
} catch (Catalog.DatabaseNotExistException e) {
throw new RuntimeException(format("database not exists: '%s'", table.getSchemaName()));
} catch (Catalog.TableAlreadyExistException e) {
Expand Down Expand Up @@ -600,7 +606,8 @@ public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(

@Override
public void addColumn(
ConnectorSession session, ConnectorTableHandle tableHandle, ColumnMetadata column) {
ConnectorSession session, ConnectorTableHandle tableHandle, ColumnMetadata column, ColumnPosition position)
{
TrinoTableHandle trinoTableHandle = (TrinoTableHandle) tableHandle;
Identifier identifier =
new Identifier(trinoTableHandle.getSchemaName(), trinoTableHandle.getTableName());
Expand Down Expand Up @@ -669,6 +676,7 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(
new ConstraintApplicationResult<>(
trinoTableHandle.copy(trinoFilter.getFilter()),
trinoFilter.getRemainFilter(),
constraint.getExpression(),
false));
} else {
return Optional.empty();
Expand Down
19 changes: 7 additions & 12 deletions src/main/java/org/apache/paimon/trino/TrinoMetadataFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,15 +18,11 @@

package org.apache.paimon.trino;

import org.apache.paimon.options.Options;
import org.apache.paimon.trino.catalog.TrinoCatalog;

import com.google.inject.Inject;
import io.trino.filesystem.TrinoFileSystemFactory;
import io.trino.hdfs.ConfigurationUtils;
import io.trino.hdfs.HdfsConfig;
import io.trino.hdfs.HdfsConfigurationInitializer;
import org.apache.hadoop.conf.Configuration;
import org.apache.paimon.options.Options;
import org.apache.paimon.trino.catalog.TrinoCatalog;

/** A factory to create {@link TrinoMetadata}. */
public class TrinoMetadataFactory {
Expand All @@ -36,14 +32,13 @@ public class TrinoMetadataFactory {
@Inject
public TrinoMetadataFactory(
Options options,
HdfsConfigurationInitializer hdfsConfigurationInitializer,
HdfsConfig hdfsConfig,
TrinoFileSystemFactory fileSystemFactory) {
Configuration configuration = null;
if (!hdfsConfig.getResourceConfigFiles().isEmpty()) {
configuration = ConfigurationUtils.getInitialConfiguration();
hdfsConfigurationInitializer.initializeConfiguration(configuration);

String confDir = options.get("hadoop-conf-dir");
if (confDir != null) {
System.setProperty("HADOOP_CONF_DIR", confDir);
}
Configuration configuration = new Configuration();

this.catalog = new TrinoCatalog(options, configuration, fileSystemFactory);
}
Expand Down
44 changes: 19 additions & 25 deletions src/main/java/org/apache/paimon/trino/TrinoPageSourceProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,23 +18,6 @@

package org.apache.paimon.trino;

import org.apache.paimon.CoreOptions;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.fileindex.FileIndexPredicate;
import org.apache.paimon.fs.Path;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.IndexFile;
import org.apache.paimon.table.source.RawFile;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.trino.catalog.TrinoCatalog;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;

import com.google.inject.Inject;
import io.airlift.units.DataSize;
import io.trino.filesystem.Location;
Expand All @@ -48,7 +31,7 @@
import io.trino.orc.OrcReaderOptions;
import io.trino.orc.OrcRecordReader;
import io.trino.orc.TupleDomainOrcPredicate;
import io.trino.plugin.hive.FileFormatDataSourceStats;
import io.trino.plugin.base.metrics.FileFormatDataSourceStats;
import io.trino.plugin.hive.orc.OrcPageSource;
import io.trino.spi.connector.ColumnHandle;
import io.trino.spi.connector.ConnectorPageSource;
Expand All @@ -61,6 +44,22 @@
import io.trino.spi.predicate.Domain;
import io.trino.spi.predicate.TupleDomain;
import io.trino.spi.type.Type;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.fileindex.FileIndexPredicate;
import org.apache.paimon.fs.Path;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.IndexFile;
import org.apache.paimon.table.source.RawFile;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.trino.catalog.TrinoCatalog;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.joda.time.DateTimeZone;

import java.io.IOException;
Expand Down Expand Up @@ -381,15 +380,12 @@ private ConnectorPageSource createOrcDataPageSource(
fileColumns.forEach(column -> fieldsMap.put(column.getColumnName(), column));
TupleDomainOrcPredicate.TupleDomainOrcPredicateBuilder predicateBuilder =
TupleDomainOrcPredicate.builder();
List<OrcPageSource.ColumnAdaptation> columnAdaptations = new ArrayList<>();
List<OrcColumn> fileReadColumns = new ArrayList<>(columns.size());
List<Type> fileReadTypes = new ArrayList<>(columns.size());

for (int i = 0; i < columns.size(); i++) {
if (columns.get(i) != null) {
// column exists
columnAdaptations.add(
OrcPageSource.ColumnAdaptation.sourceColumn(fileReadColumns.size()));
OrcColumn orcColumn = fieldsMap.get(columns.get(i));
if (orcColumn == null) {
throw new RuntimeException(
Expand All @@ -400,8 +396,6 @@ private ConnectorPageSource createOrcDataPageSource(
if (domains.get(i) != null) {
predicateBuilder.addColumn(orcColumn.getColumnId(), domains.get(i));
}
} else {
columnAdaptations.add(OrcPageSource.ColumnAdaptation.nullColumn(types.get(i)));
}
}

Expand All @@ -410,15 +404,15 @@ private ConnectorPageSource createOrcDataPageSource(
reader.createRecordReader(
fileReadColumns,
fileReadTypes,
false,
predicateBuilder.build(),
DateTimeZone.UTC,
memoryUsage,
INITIAL_BATCH_SIZE,
RuntimeException::new);
exception -> new RuntimeException("Error reading ORC file", exception));

return new OrcPageSource(
recordReader,
columnAdaptations,
orcDataSource,
Optional.empty(),
Optional.empty(),
Expand Down
Loading