You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "leaves12138 (via GitHub)" <gi...@apache.org> on 2024/03/13 09:25:09 UTC
[PR] Refactory paimon trino module, use `TrinoFileSystem` as the only file system. [incubator-paimon-trino]
leaves12138 opened a new pull request, #54:
URL: https://github.com/apache/incubator-paimon-trino/pull/54
This pull request has done three things:
1. Use trino-filesystem as paimon file io base in 427
2. Use trino filesystem to read file directly if possible
3. Do not support old version, use 427 version as common
Removed versions before 422. No support for them any more.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactory paimon trino module, use `TrinoFileSystem` as the only file system. [incubator-paimon-trino]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #54:
URL: https://github.com/apache/incubator-paimon-trino/pull/54#issuecomment-1994088611
Really thanks @leaves12138 ! Left some minor comments.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactory paimon trino module, use `TrinoFileSystem` as the only file system. [incubator-paimon-trino]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi closed pull request #54: Refactory paimon trino module, use `TrinoFileSystem` as the only file system.
URL: https://github.com/apache/incubator-paimon-trino/pull/54
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactory paimon trino module, use `TrinoFileSystem` as the only file system. [incubator-paimon-trino]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #54:
URL: https://github.com/apache/incubator-paimon-trino/pull/54#discussion_r1522976926
##########
paimon-trino-422/src/main/java/org/apache/paimon/trino/fileio/TrinoFileIO.java:
##########
@@ -0,0 +1,164 @@
+/*
+ *
+ * * 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.paimon.trino.fileio;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+
+import io.trino.filesystem.FileEntry;
+import io.trino.filesystem.FileIterator;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoInputFile;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Trino file io for paimon. */
+public class TrinoFileIO implements FileIO {
+
+ private final TrinoFileSystem trinoFileSystem;
+ private final boolean objectStore;
+
+ public TrinoFileIO(TrinoFileSystem trinoFileSystem, @Nullable Path path) {
+ this.trinoFileSystem = trinoFileSystem;
+ this.objectStore = path == null || checkObjectStore(path.toUri().getScheme());
+ }
+
+ @Override
+ public boolean isObjectStore() {
+ return objectStore;
+ }
+
+ @Override
+ public void configure(CatalogContext catalogContext) {}
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws IOException {
+ return new TrinoInputStreamWrapper(
+ trinoFileSystem.newInputFile(Location.of(path.toString())).newStream());
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean b) throws IOException {
+ return new PositionOutputStreamWrapper(
+ trinoFileSystem.newOutputFile(Location.of(path.toString())).create());
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ return status(path);
+ }
+
+ private FileStatus status(Path path) throws IOException {
+ if (trinoFileSystem.directoryExists(Location.of(path.toString())).orElse(false)) {
+ return new TrinoDirectoryFileStatus(path);
+ } else {
+ TrinoInputFile trinoInputFile =
+ trinoFileSystem.newInputFile(Location.of(path.toString()));
+ return new TrinoFileStatus(
+ trinoInputFile.length(), path, trinoInputFile.lastModified().getEpochSecond());
+ }
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ List<FileStatus> fileStatusList = new ArrayList<>();
+ Location location = Location.of(path.toString());
+ if (trinoFileSystem.directoryExists(location).orElse(false)) {
+ FileIterator fileIterator = trinoFileSystem.listFiles(location);
+ while (fileIterator.hasNext()) {
+ FileEntry fileEntry = fileIterator.next();
+ fileStatusList.add(
+ new TrinoFileStatus(
+ fileEntry.length(),
+ new Path(fileEntry.location().toString()),
+ fileEntry.lastModified().getEpochSecond()));
+ }
+ }
Review Comment:
add comment here
##########
paimon-trino-common/src/main/java/org/apache/paimon/trino/TrinoPageSourceProvider.java:
##########
@@ -64,32 +113,229 @@ public ConnectorPageSource createPageSource(
}
private ConnectorPageSource createPageSource(
+ ConnectorSession session,
Table table,
TupleDomain<TrinoColumnHandle> filter,
TrinoSplit split,
List<ColumnHandle> columns,
OptionalLong limit) {
- ReadBuilder read = table.newReadBuilder();
RowType rowType = table.rowType();
List<String> fieldNames = FieldNameUtils.fieldNames(rowType);
List<String> projectedFields =
columns.stream()
.map(TrinoColumnHandle.class::cast)
.map(TrinoColumnHandle::getColumnName)
- .collect(Collectors.toList());
- if (!fieldNames.equals(projectedFields)) {
- int[] projected = projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
- read.withProjection(projected);
+ .toList();
+ int[] columnIndex =
+ // the column index, very important
+ projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
+
+ TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+
+ try {
+ Split paimonSplit = split.decodeSplit();
+ Optional<List<RawFile>> optionalRawFiles = paimonSplit.convertToRawFiles();
+ if (checkRawFile(optionalRawFiles)) {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ SchemaManager schemaManager =
+ new SchemaManager(fileStoreTable.fileIO(), fileStoreTable.location());
+ List<Type> type =
+ columns.stream()
+ .map(s -> ((TrinoColumnHandle) s).getTrinoType())
+ .collect(Collectors.toList());
+ try {
+ return new DirectTrinoPageSource(
+ optionalRawFiles.orElseThrow().stream()
+ .map(
+ rawFile ->
+ createDataPageSource(
+ rawFile.format(),
+ fileSystem.newInputFile(
+ Location.of(rawFile.path())),
+ fileStoreTable.coreOptions(),
+ // map table column index to data column
+ // index, if column does not exist in
+ // data columns, set it to -1
+ // columns those set to -1 will generate
+ // a null vector in orc page
+ mapping(
+ columnIndex,
+ rowType.getFields(),
+ schemaManager
+ .schema(
+ rawFile
+ .schemaId())
+ .fields()),
+ type,
+ orderDomains(projectedFields, filter)))
+ .collect(
+ Collector.of(
+ LinkedList::new,
+ List::add,
+ (left, right) -> {
+ left.addAll(right);
+ return left;
+ })));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ // old read way
+ ReadBuilder read = table.newReadBuilder();
+ new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+
+ if (!fieldNames.equals(projectedFields)) {
+ read.withProjection(columnIndex);
+ }
+
+ return new TrinoPageSource(
+ read.newRead().executeFilter().createReader(paimonSplit), columns, limit);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
+ }
+
+ // make domains(filters) to be ordered by projected fields' order.
+ private List<Domain> orderDomains(
+ List<String> projectedFields, TupleDomain<TrinoColumnHandle> filter) {
+ Optional<Map<TrinoColumnHandle, Domain>> optionalFilter = filter.getDomains();
+ Map<String, Domain> domainMap = new HashMap<>();
+ optionalFilter.ifPresent(
+ trinoColumnHandleDomainMap ->
+ trinoColumnHandleDomainMap.forEach(
+ (k, v) -> domainMap.put(k.getColumnName(), v)));
- new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+ return projectedFields.stream()
+ .map(name -> domainMap.getOrDefault(name, null))
+ .collect(Collectors.toList());
+ }
+
+ private boolean checkRawFile(Optional<List<RawFile>> optionalRawFiles) {
+ return optionalRawFiles.isPresent() && checkFormat(optionalRawFiles.get());
+ }
+ // only support orc yet.
+ // todo: support parquet and avro
Review Comment:
todo => TODO
##########
paimon-trino-common/src/main/java/org/apache/paimon/trino/TrinoPageSourceProvider.java:
##########
@@ -64,32 +113,229 @@ public ConnectorPageSource createPageSource(
}
private ConnectorPageSource createPageSource(
+ ConnectorSession session,
Table table,
TupleDomain<TrinoColumnHandle> filter,
TrinoSplit split,
List<ColumnHandle> columns,
OptionalLong limit) {
- ReadBuilder read = table.newReadBuilder();
RowType rowType = table.rowType();
List<String> fieldNames = FieldNameUtils.fieldNames(rowType);
List<String> projectedFields =
columns.stream()
.map(TrinoColumnHandle.class::cast)
.map(TrinoColumnHandle::getColumnName)
- .collect(Collectors.toList());
- if (!fieldNames.equals(projectedFields)) {
- int[] projected = projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
- read.withProjection(projected);
+ .toList();
+ int[] columnIndex =
+ // the column index, very important
+ projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
+
+ TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+
+ try {
+ Split paimonSplit = split.decodeSplit();
+ Optional<List<RawFile>> optionalRawFiles = paimonSplit.convertToRawFiles();
+ if (checkRawFile(optionalRawFiles)) {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ SchemaManager schemaManager =
+ new SchemaManager(fileStoreTable.fileIO(), fileStoreTable.location());
+ List<Type> type =
+ columns.stream()
+ .map(s -> ((TrinoColumnHandle) s).getTrinoType())
+ .collect(Collectors.toList());
+ try {
+ return new DirectTrinoPageSource(
+ optionalRawFiles.orElseThrow().stream()
+ .map(
+ rawFile ->
+ createDataPageSource(
+ rawFile.format(),
+ fileSystem.newInputFile(
+ Location.of(rawFile.path())),
+ fileStoreTable.coreOptions(),
+ // map table column index to data column
+ // index, if column does not exist in
+ // data columns, set it to -1
+ // columns those set to -1 will generate
+ // a null vector in orc page
+ mapping(
+ columnIndex,
+ rowType.getFields(),
+ schemaManager
+ .schema(
+ rawFile
+ .schemaId())
+ .fields()),
+ type,
+ orderDomains(projectedFields, filter)))
+ .collect(
+ Collector.of(
+ LinkedList::new,
+ List::add,
+ (left, right) -> {
+ left.addAll(right);
+ return left;
+ })));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ // old read way
+ ReadBuilder read = table.newReadBuilder();
+ new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+
+ if (!fieldNames.equals(projectedFields)) {
+ read.withProjection(columnIndex);
+ }
+
+ return new TrinoPageSource(
+ read.newRead().executeFilter().createReader(paimonSplit), columns, limit);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
+ }
+
+ // make domains(filters) to be ordered by projected fields' order.
+ private List<Domain> orderDomains(
+ List<String> projectedFields, TupleDomain<TrinoColumnHandle> filter) {
+ Optional<Map<TrinoColumnHandle, Domain>> optionalFilter = filter.getDomains();
+ Map<String, Domain> domainMap = new HashMap<>();
+ optionalFilter.ifPresent(
+ trinoColumnHandleDomainMap ->
+ trinoColumnHandleDomainMap.forEach(
+ (k, v) -> domainMap.put(k.getColumnName(), v)));
- new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+ return projectedFields.stream()
+ .map(name -> domainMap.getOrDefault(name, null))
+ .collect(Collectors.toList());
+ }
+
+ private boolean checkRawFile(Optional<List<RawFile>> optionalRawFiles) {
+ return optionalRawFiles.isPresent() && checkFormat(optionalRawFiles.get());
+ }
+ // only support orc yet.
+ // todo: support parquet and avro
+ private boolean checkFormat(List<RawFile> rawFiles) {
+ for (RawFile rawFile : rawFiles) {
+ if (!rawFile.format().equals("orc")) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ // map the table schema columnsIndex to data schema columnsIndex
+ private int[] mapping(
+ int[] tableSchemaColumnIndex, List<DataField> tableFields, List<DataField> dataFields) {
+
+ int[] mapping = createIndexMapping(tableFields, dataFields);
+ if (mapping == null) {
+ return tableSchemaColumnIndex;
+ }
+ int[] result = new int[tableSchemaColumnIndex.length];
+
+ for (int i = 0; i < tableSchemaColumnIndex.length; i++) {
+ int po = tableSchemaColumnIndex[i];
+ result[i] = mapping[po];
+ }
+ return result;
+ }
+
+ private ConnectorPageSource createDataPageSource(
+ String format,
+ TrinoInputFile inputFile,
+ // todo construct read option by core-options
+ CoreOptions coreOptions,
+ int[] columns,
+ List<Type> types,
+ List<Domain> domains) {
+ switch (format) {
+ case "orc":
+ {
+ return createOrcDataPageSource(
+ inputFile,
+ new OrcReaderOptions()
+ .withTinyStripeThreshold(
Review Comment:
Why set tiny tripe: add comment to explain.
##########
paimon-trino-422/src/main/java/org/apache/paimon/trino/fileio/TrinoFileIO.java:
##########
@@ -0,0 +1,164 @@
+/*
+ *
+ * * 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.paimon.trino.fileio;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+
+import io.trino.filesystem.FileEntry;
+import io.trino.filesystem.FileIterator;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoInputFile;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Trino file io for paimon. */
+public class TrinoFileIO implements FileIO {
+
+ private final TrinoFileSystem trinoFileSystem;
+ private final boolean objectStore;
+
+ public TrinoFileIO(TrinoFileSystem trinoFileSystem, @Nullable Path path) {
+ this.trinoFileSystem = trinoFileSystem;
+ this.objectStore = path == null || checkObjectStore(path.toUri().getScheme());
+ }
+
+ @Override
+ public boolean isObjectStore() {
+ return objectStore;
+ }
+
+ @Override
+ public void configure(CatalogContext catalogContext) {}
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws IOException {
+ return new TrinoInputStreamWrapper(
+ trinoFileSystem.newInputFile(Location.of(path.toString())).newStream());
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(Path path, boolean b) throws IOException {
+ return new PositionOutputStreamWrapper(
+ trinoFileSystem.newOutputFile(Location.of(path.toString())).create());
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ return status(path);
+ }
+
+ private FileStatus status(Path path) throws IOException {
+ if (trinoFileSystem.directoryExists(Location.of(path.toString())).orElse(false)) {
+ return new TrinoDirectoryFileStatus(path);
+ } else {
+ TrinoInputFile trinoInputFile =
+ trinoFileSystem.newInputFile(Location.of(path.toString()));
+ return new TrinoFileStatus(
+ trinoInputFile.length(), path, trinoInputFile.lastModified().getEpochSecond());
+ }
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ List<FileStatus> fileStatusList = new ArrayList<>();
+ Location location = Location.of(path.toString());
+ if (trinoFileSystem.directoryExists(location).orElse(false)) {
+ FileIterator fileIterator = trinoFileSystem.listFiles(location);
+ while (fileIterator.hasNext()) {
+ FileEntry fileEntry = fileIterator.next();
+ fileStatusList.add(
+ new TrinoFileStatus(
+ fileEntry.length(),
+ new Path(fileEntry.location().toString()),
+ fileEntry.lastModified().getEpochSecond()));
+ }
+ }
+ return fileStatusList.toArray(new FileStatus[0]);
+ }
+
+ @Override
+ public boolean exists(Path path) throws IOException {
+ return trinoFileSystem.directoryExists(Location.of(path.toString())).orElse(false)
+ || existFile(Location.of(path.toString()));
+ }
+
+ private boolean existFile(Location location) throws IOException {
+ try {
+ return trinoFileSystem.newInputFile(location).exists();
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
+ }
+
+ @Override
+ public boolean delete(Path path, boolean b) throws IOException {
+ Location location = Location.of(path.toString());
+ if (trinoFileSystem.directoryExists(location).orElse(false)) {
+ trinoFileSystem.deleteDirectory(location);
+ return true;
+ } else if (existFile(location)) {
+ trinoFileSystem.deleteFile(location);
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public boolean mkdirs(Path path) throws IOException {
+ throw new UnsupportedEncodingException();
Review Comment:
UnsupportedOperationException
##########
paimon-trino-common/src/main/java/org/apache/paimon/trino/TrinoPageSourceProvider.java:
##########
@@ -64,32 +113,229 @@ public ConnectorPageSource createPageSource(
}
private ConnectorPageSource createPageSource(
+ ConnectorSession session,
Table table,
TupleDomain<TrinoColumnHandle> filter,
TrinoSplit split,
List<ColumnHandle> columns,
OptionalLong limit) {
- ReadBuilder read = table.newReadBuilder();
RowType rowType = table.rowType();
List<String> fieldNames = FieldNameUtils.fieldNames(rowType);
List<String> projectedFields =
columns.stream()
.map(TrinoColumnHandle.class::cast)
.map(TrinoColumnHandle::getColumnName)
- .collect(Collectors.toList());
- if (!fieldNames.equals(projectedFields)) {
- int[] projected = projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
- read.withProjection(projected);
+ .toList();
+ int[] columnIndex =
+ // the column index, very important
+ projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
+
+ TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+
+ try {
+ Split paimonSplit = split.decodeSplit();
+ Optional<List<RawFile>> optionalRawFiles = paimonSplit.convertToRawFiles();
+ if (checkRawFile(optionalRawFiles)) {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ SchemaManager schemaManager =
+ new SchemaManager(fileStoreTable.fileIO(), fileStoreTable.location());
+ List<Type> type =
+ columns.stream()
+ .map(s -> ((TrinoColumnHandle) s).getTrinoType())
+ .collect(Collectors.toList());
+ try {
+ return new DirectTrinoPageSource(
+ optionalRawFiles.orElseThrow().stream()
+ .map(
+ rawFile ->
+ createDataPageSource(
+ rawFile.format(),
+ fileSystem.newInputFile(
+ Location.of(rawFile.path())),
+ fileStoreTable.coreOptions(),
+ // map table column index to data column
+ // index, if column does not exist in
+ // data columns, set it to -1
+ // columns those set to -1 will generate
+ // a null vector in orc page
+ mapping(
+ columnIndex,
+ rowType.getFields(),
+ schemaManager
+ .schema(
+ rawFile
+ .schemaId())
+ .fields()),
+ type,
+ orderDomains(projectedFields, filter)))
+ .collect(
+ Collector.of(
+ LinkedList::new,
+ List::add,
+ (left, right) -> {
+ left.addAll(right);
+ return left;
+ })));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ // old read way
+ ReadBuilder read = table.newReadBuilder();
+ new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+
+ if (!fieldNames.equals(projectedFields)) {
+ read.withProjection(columnIndex);
+ }
+
+ return new TrinoPageSource(
+ read.newRead().executeFilter().createReader(paimonSplit), columns, limit);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
+ }
+
+ // make domains(filters) to be ordered by projected fields' order.
+ private List<Domain> orderDomains(
+ List<String> projectedFields, TupleDomain<TrinoColumnHandle> filter) {
+ Optional<Map<TrinoColumnHandle, Domain>> optionalFilter = filter.getDomains();
+ Map<String, Domain> domainMap = new HashMap<>();
+ optionalFilter.ifPresent(
+ trinoColumnHandleDomainMap ->
+ trinoColumnHandleDomainMap.forEach(
+ (k, v) -> domainMap.put(k.getColumnName(), v)));
- new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+ return projectedFields.stream()
+ .map(name -> domainMap.getOrDefault(name, null))
+ .collect(Collectors.toList());
+ }
+
+ private boolean checkRawFile(Optional<List<RawFile>> optionalRawFiles) {
+ return optionalRawFiles.isPresent() && checkFormat(optionalRawFiles.get());
+ }
+ // only support orc yet.
+ // todo: support parquet and avro
+ private boolean checkFormat(List<RawFile> rawFiles) {
+ for (RawFile rawFile : rawFiles) {
+ if (!rawFile.format().equals("orc")) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ // map the table schema columnsIndex to data schema columnsIndex
+ private int[] mapping(
+ int[] tableSchemaColumnIndex, List<DataField> tableFields, List<DataField> dataFields) {
+
+ int[] mapping = createIndexMapping(tableFields, dataFields);
+ if (mapping == null) {
+ return tableSchemaColumnIndex;
+ }
+ int[] result = new int[tableSchemaColumnIndex.length];
+
+ for (int i = 0; i < tableSchemaColumnIndex.length; i++) {
+ int po = tableSchemaColumnIndex[i];
+ result[i] = mapping[po];
+ }
+ return result;
+ }
+
+ private ConnectorPageSource createDataPageSource(
+ String format,
+ TrinoInputFile inputFile,
+ // todo construct read option by core-options
+ CoreOptions coreOptions,
+ int[] columns,
+ List<Type> types,
+ List<Domain> domains) {
+ switch (format) {
+ case "orc":
+ {
+ return createOrcDataPageSource(
+ inputFile,
+ new OrcReaderOptions()
Review Comment:
Add a TODO: pass options from catalog configuration.
##########
paimon-trino-common/src/main/java/org/apache/paimon/trino/TrinoPageSourceProvider.java:
##########
@@ -64,32 +113,229 @@ public ConnectorPageSource createPageSource(
}
private ConnectorPageSource createPageSource(
+ ConnectorSession session,
Table table,
TupleDomain<TrinoColumnHandle> filter,
TrinoSplit split,
List<ColumnHandle> columns,
OptionalLong limit) {
- ReadBuilder read = table.newReadBuilder();
RowType rowType = table.rowType();
List<String> fieldNames = FieldNameUtils.fieldNames(rowType);
List<String> projectedFields =
columns.stream()
.map(TrinoColumnHandle.class::cast)
.map(TrinoColumnHandle::getColumnName)
- .collect(Collectors.toList());
- if (!fieldNames.equals(projectedFields)) {
- int[] projected = projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
- read.withProjection(projected);
+ .toList();
+ int[] columnIndex =
+ // the column index, very important
+ projectedFields.stream().mapToInt(fieldNames::indexOf).toArray();
+
+ TrinoFileSystem fileSystem = fileSystemFactory.create(session);
+
+ try {
+ Split paimonSplit = split.decodeSplit();
+ Optional<List<RawFile>> optionalRawFiles = paimonSplit.convertToRawFiles();
+ if (checkRawFile(optionalRawFiles)) {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ SchemaManager schemaManager =
+ new SchemaManager(fileStoreTable.fileIO(), fileStoreTable.location());
+ List<Type> type =
+ columns.stream()
+ .map(s -> ((TrinoColumnHandle) s).getTrinoType())
+ .collect(Collectors.toList());
+ try {
+ return new DirectTrinoPageSource(
+ optionalRawFiles.orElseThrow().stream()
+ .map(
+ rawFile ->
+ createDataPageSource(
+ rawFile.format(),
+ fileSystem.newInputFile(
+ Location.of(rawFile.path())),
+ fileStoreTable.coreOptions(),
+ // map table column index to data column
+ // index, if column does not exist in
+ // data columns, set it to -1
+ // columns those set to -1 will generate
+ // a null vector in orc page
+ mapping(
+ columnIndex,
+ rowType.getFields(),
+ schemaManager
+ .schema(
+ rawFile
+ .schemaId())
+ .fields()),
+ type,
+ orderDomains(projectedFields, filter)))
+ .collect(
+ Collector.of(
+ LinkedList::new,
+ List::add,
+ (left, right) -> {
+ left.addAll(right);
+ return left;
+ })));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ // old read way
+ ReadBuilder read = table.newReadBuilder();
+ new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+
+ if (!fieldNames.equals(projectedFields)) {
+ read.withProjection(columnIndex);
+ }
+
+ return new TrinoPageSource(
+ read.newRead().executeFilter().createReader(paimonSplit), columns, limit);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
+ }
+
+ // make domains(filters) to be ordered by projected fields' order.
+ private List<Domain> orderDomains(
+ List<String> projectedFields, TupleDomain<TrinoColumnHandle> filter) {
+ Optional<Map<TrinoColumnHandle, Domain>> optionalFilter = filter.getDomains();
+ Map<String, Domain> domainMap = new HashMap<>();
+ optionalFilter.ifPresent(
+ trinoColumnHandleDomainMap ->
+ trinoColumnHandleDomainMap.forEach(
+ (k, v) -> domainMap.put(k.getColumnName(), v)));
- new TrinoFilterConverter(rowType).convert(filter).ifPresent(read::withFilter);
+ return projectedFields.stream()
+ .map(name -> domainMap.getOrDefault(name, null))
+ .collect(Collectors.toList());
+ }
+
+ private boolean checkRawFile(Optional<List<RawFile>> optionalRawFiles) {
Review Comment:
check like here will throw exception.
You can rename it to `canUseTrinoPageSource`
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Refactory paimon trino module, use `TrinoFileSystem` as the only file system. [incubator-paimon-trino]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #54:
URL: https://github.com/apache/incubator-paimon-trino/pull/54
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org