Port OrcFileWriterFactory.createOrcDataSink to TrinoFileSystem #14627

homar · 2022-10-13T23:22:19Z

Description

This is prerequisite to be able to track FileSystem memory utilization while writing ORC data - #14023

Non-technical explanation

Allows better tracking of memory consumed internally.

Release notes

(x) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix some things. ({issue}`issuenumber`)

homar · 2022-10-14T07:40:20Z

CI: #14631

sopel39 · 2022-10-14T15:32:36Z

This is prerequisite to be able to track FileSystem memory utilization while writing ORC or Parquet data

homar · 2022-10-14T16:24:08Z

Sorry I didn't think anyone is going to review it at this stage.

alexjo2144 · 2022-10-17T08:34:28Z

Sorry I didn't think anyone is going to review it at this stage.

I am not going to review it then, but you can @ mention me when you want me to take a look

sopel39 · 2022-10-17T08:35:35Z

Yup. Please mark it ready-for review and re-request review when this is ready

homar · 2022-10-17T10:51:43Z

@alexjo2144 @sopel39 there is no button for re-requesting the review but please take a look ;)

sopel39

mostly lgtm

sopel39 · 2022-10-21T13:57:09Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HivePageSinkProvider.java

@@ -58,6 +59,7 @@
        implements ConnectorPageSinkProvider
 {
    private final Set<HiveFileWriterFactory> fileWriterFactories;
+    private final TrinoFileSystemFactory trinoFileSystemFactory;


just fileSystemFactory

sopel39 · 2022-10-21T13:58:15Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HivePageSinkProvider.java

@@ -79,6 +81,7 @@
    @Inject
    public HivePageSinkProvider(
            Set<HiveFileWriterFactory> fileWriterFactories,
+            TrinoFileSystemFactory trinoFileSystemFactory,


just fileSystemFactory

sopel39 · 2022-10-21T13:58:23Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HivePageSinkProvider.java

@@ -93,6 +96,7 @@ public HivePageSinkProvider(
            HiveWriterStats hiveWriterStats)
    {
        this.fileWriterFactories = ImmutableSet.copyOf(requireNonNull(fileWriterFactories, "fileWriterFactories is null"));
+        this.trinoFileSystemFactory = requireNonNull(trinoFileSystemFactory, "trinoFileSystemFactory is null");


just fileSystemFactory here and in other places

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

sopel39 · 2022-10-21T14:08:04Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/SortingFileWriter.java

@@ -212,12 +213,13 @@ private void mergeFiles(Iterable<TempFile> files, Consumer<Page> consumer)
            Collection<Iterator<Page>> iterators = new ArrayList<>();

            for (TempFile tempFile : files) {
-                Path file = tempFile.getPath();
+                String file = tempFile.getPath();
+                TrinoInputFile trinoInputFile = trinoFileSystem.newInputFile(file);


trinoInputFile -> inputFile

sopel39 · 2022-10-21T14:22:42Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcDeleteDeltaPageSource.java

-            FileSystem fileSystem = hdfsEnvironment.getFileSystem(identity, path, configuration);
-            FSDataInputStream inputStream = hdfsEnvironment.doAs(identity, () -> fileSystem.open(path));
+            TrinoFileSystem fileSystem = trinoFileSystemFactory.create(identity);
+            TrinoInput inputStream = hdfsEnvironment.doAs(identity, () -> fileSystem.newInputFile(path).newInput());


you don't need to wrap in doAs anymore. See io.trino.filesystem.hdfs.HdfsInputFile#newInput

sopel39 · 2022-10-21T14:24:54Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcFileWriterFactory.java


            Optional<Supplier<OrcDataSource>> validationInputFactory = Optional.empty();
+            String stringPath = path.toString();


if you move it higher, you can also use it in createOrcDataSink(trinoFileSystem, path.toString());

sopel39 · 2022-10-21T14:25:45Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcPageSourceFactory.java

-            FileSystem fileSystem = hdfsEnvironment.getFileSystem(identity, path, configuration);
-            FSDataInputStream inputStream = hdfsEnvironment.doAs(identity, () -> fileSystem.open(path));
+            TrinoFileSystem trinoFileSystem = trinoFileSystemFactory.create(identity);
+            TrinoInput inputStream = hdfsEnvironment.doAs(identity, () -> trinoFileSystem.newInputFile(path.toString()).newInput());


doAs is no longer needed here

sopel39 · 2022-10-21T14:26:29Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OriginalFilesUtils.java

-            FileSystem fileSystem = hdfsEnvironment.getFileSystem(identity, splitPath, configuration);
-            FSDataInputStream inputStream = hdfsEnvironment.doAs(identity, () -> fileSystem.open(splitPath));
+            TrinoFileSystem trinoFileSystem = trinoFileSystemFactory.create(identity);
+            TrinoInput trinoInput = hdfsEnvironment.doAs(identity, () -> trinoFileSystem.newInputFile(splitPath).newInput());


trinoInput - > inputFile would be better (here and in other places)

but that is not a file it is more like inputStream, see io.trino.filesystem.TrinoInput

sopel39 · 2022-10-21T14:28:07Z

plugin/trino-hive/src/test/java/io/trino/plugin/hive/benchmark/StandardFileFormats.java

@@ -143,7 +143,7 @@ public HiveStorageFormat getFormat()
        @Override
        public Optional<HivePageSourceFactory> getHivePageSourceFactory(HdfsEnvironment hdfsEnvironment)
        {
-            return Optional.of(new OrcPageSourceFactory(new OrcReaderOptions(), hdfsEnvironment, new FileFormatDataSourceStats(), UTC));
+            return Optional.of(new OrcPageSourceFactory(new OrcReaderOptions(), hdfsEnvironment, new FileFormatDataSourceStats(), UTC, new HdfsFileSystemFactory(hdfsEnvironment)));


why not use HDFS_FILE_SYSTEM_FACTORY?

homar · 2022-10-24T15:30:48Z

CI: #14686

sopel39 · 2022-10-26T09:41:26Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

@@ -636,22 +638,34 @@ public HiveWriter createWriter(Page partitionColumns, int position, OptionalInt
                hiveWriterStats);
    }

+    private String setSchemeToFileIfOneDoesNotExist(String path)


move below makeRowIdSortingWriter

sopel39 · 2022-10-26T09:42:28Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

@@ -636,22 +638,34 @@ public HiveWriter createWriter(Page partitionColumns, int position, OptionalInt
                hiveWriterStats);
    }



I think you could add unit test for it in io.trino.plugin.hive.TestHiveWriterFactory if you make this method staic and package private (then move this method to bottom of class)

sopel39 · 2022-10-26T09:42:51Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

+        Path parent = path.getParent();
+        String scheme = parent.toUri().getScheme();
+        String parentPath = parent.toString();
+        if (scheme == null || scheme.equals("")) {


use setSchemeToFileIfOneDoesNotExist

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

sopel39

lgtm % comments

sopel39 · 2022-10-26T11:20:42Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

            Path tempFilePath;
            if (sortedWritingTempStagingPathEnabled) {
                String tempPrefix = sortedWritingTempStagingPath.replace(
                        "${USER}",
                        new HdfsContext(session).getIdentity().getUser());
+                tempPrefix = setSchemeToFileIfOneDoesNotExist(tempPrefix);


maybe setSchemeToFileIfAbsent

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcDeletedRows.java

sopel39 · 2022-10-26T11:30:02Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/HdfsOrcDataSource.java

    private final FileFormatDataSourceStats stats;

    public HdfsOrcDataSource(
            OrcDataSourceId id,
            long size,
            OrcReaderOptions options,
-            FSDataInputStream inputStream,
+            TrinoInput input,


It would be cleaner if this was TrinoInputFile. Then HdfsOrcDataSource would really own lifecycle of TrinoInput (both create it and close it).

sopel39 · 2022-10-26T11:30:54Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcFileWriterFactory.java

                                new OrcReaderOptions(),
-                                fileSystem.open(path),
+                                inputFile.newInput(),


Can HdfsOrcDataSource just accept inputFile?

alexjo2144 · 2022-10-26T15:35:16Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

            }
-            catch (IOException e) {
+            catch (RuntimeException e) {


If it doesn't throw anything just remove the try/catch

I don't want to change when specific exception is thrown, if you check deeper then you will see that I cach IOException there and throw UncheckedIO. I will change this to catch UncheckedIO only. Does it make sense ?

alexjo2144 · 2022-10-26T15:36:07Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

        }
-        catch (IOException e) {
+        catch (RuntimeException e) {


same as above

alexjo2144 · 2022-10-26T15:37:25Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/SortingFileWriter.java

+                String file = tempFile.getPath();
+                fileSystem.deleteFile(file);


Super nit: I would just inline this

Suggested change

String file = tempFile.getPath();

fileSystem.deleteFile(file);

fileSystem.deleteFile(tempFile.getPath());

alexjo2144 · 2022-10-26T15:39:37Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcDeleteDeltaPageSource.java

@@ -139,7 +137,7 @@ private OrcDeleteDeltaPageSource(
        this.stats = requireNonNull(stats, "stats is null");
        this.orcDataSource = requireNonNull(orcDataSource, "orcDataSource is null");

-        verifyAcidSchema(reader, path);
+        verifyAcidSchema(reader, new Path(path));


You could migrate verifyAcidSchema pretty easily as well. Looks like it just uses the path in error messages

I actually tried that and it turned out it is not that simple - I would also have to make changes in OrcPageSourceFactory to verifyFileHasColumnNames and some more places - there would be a lot of changes and this is not really related.

alexjo2144 · 2022-10-26T15:40:28Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcFileWriterFactory.java

    private final TypeManager typeManager;
    private final NodeVersion nodeVersion;
    private final FileFormatDataSourceStats readStats;
    private final OrcWriterStats stats = new OrcWriterStats();
    private final OrcWriterOptions orcWriterOptions;
+    private final TrinoFileSystemFactory fileSystemFactory;


Nit: I'd put this at the top where HdfsEnvironment was

alexjo2144 · 2022-10-26T15:42:42Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcPageSourceFactory.java

+            HdfsEnvironment hdfsEnvironment,
+            FileFormatDataSourceStats stats,
+            HiveConfig hiveConfig,
+            TrinoFileSystemFactory fileSystemFactory)


This is kinda a half-way migration right? We still need the hdfsEnvironment for some places. Let's put the two fields together in the field list

sopel39

lgtm % comments

sopel39 · 2022-10-28T09:11:20Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

@@ -289,15 +289,6 @@ public HiveWriterFactory(

        Configuration conf = hdfsEnvironment.getConfiguration(new HdfsContext(session), writePath);
        this.conf = toJobConf(conf);
-
-        // make sure the FileSystem is created with the correct Configuration object


undo. Let's keep this

sopel39 · 2022-10-28T09:12:15Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

+                            }
+                        }
+                        catch (IOException e) {
+                            throw new TrinoException(HIVE_FILESYSTEM_ERROR, format("Error while accessing: %s", writeInfoTargetPath));


propagate e

sopel39 · 2022-10-28T09:30:35Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

-        catch (IOException e) {
-            throw new TrinoException(HIVE_WRITER_OPEN_ERROR, e);
-        }
+        TrinoFileSystem fileSystem;


remove this line

sopel39 · 2022-10-28T09:30:53Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

+        String parentPath = parent.toString();
+        parentPath = setSchemeToFileIfAbsent(parentPath);
+        Path tempFilePath = new Path(parentPath, ".tmp-sort." + path.getName());
+        fileSystem = fileSystemFactory.create(session);


TrinoFileSystem fileSystem = ...

sopel39 · 2022-10-28T09:31:29Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

-        }
+        TrinoFileSystem fileSystem;
+        Path parent = path.getParent();
+        String parentPath = parent.toString();


you can inline it: String parentPath = setSchemeToFileIfAbsent(parent.toString())

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveWriterFactory.java

plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveWriterFactory.java

homar · 2022-10-28T18:56:44Z

CI: #12818

homar · 2022-10-28T23:37:37Z

CI #14814

sopel39

lgtm % comment

sopel39 · 2022-11-02T08:31:16Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcPageSourceFactory.java

@@ -123,16 +124,23 @@
    private static final Pattern DEFAULT_HIVE_COLUMN_NAME_PATTERN = Pattern.compile("_col\\d+");
    private final OrcReaderOptions orcReaderOptions;
    private final HdfsEnvironment hdfsEnvironment;


you can remove hdfsEnvironment now. It seems unused

It is still used, it is passed to io.trino.plugin.hive.orc.OrcPageSourceFactory#createOrcPageSource and then used to create OrcDeletedRows.

We could try to simplify it further

sopel39 · 2022-11-02T08:31:51Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/orc/OrcPageSourceFactory.java

@@ -252,13 +263,13 @@ private ConnectorPageSource createOrcPageSource(

        boolean originalFilesPresent = acidInfo.isPresent() && !acidInfo.get().getOriginalFiles().isEmpty();
        try {
-            FileSystem fileSystem = hdfsEnvironment.getFileSystem(identity, path, configuration);


HdfsEnvironment hdfsEnvironment is unused now

It is used on line 414

cla-bot bot added the cla-signed label Oct 13, 2022

github-actions bot added the tests:hive label Oct 13, 2022

martint force-pushed the master branch from bf8f0c9 to b8602fc Compare October 14, 2022 01:04

homar force-pushed the homar/test_using_trino_filesystem_in_trino_hive_for_orc branch from adb3467 to dd5fe8f Compare October 14, 2022 08:09

findepi requested review from sopel39, electrum and alexjo2144 October 14, 2022 15:12

homar force-pushed the homar/test_using_trino_filesystem_in_trino_hive_for_orc branch from dd18dca to 2c9cae3 Compare October 17, 2022 10:47

homar marked this pull request as ready for review October 17, 2022 10:48

findepi requested review from sopel39 and removed request for sopel39 October 18, 2022 07:59

homar mentioned this pull request Oct 21, 2022

Account for memory allocated by trino s3 staging output stream #14407

Merged

sopel39 reviewed Oct 21, 2022

View reviewed changes

homar force-pushed the homar/test_using_trino_filesystem_in_trino_hive_for_orc branch 4 times, most recently from 24a085f to 8f35ca7 Compare October 24, 2022 13:34

homar requested a review from sopel39 October 24, 2022 15:31

homar self-assigned this Oct 25, 2022

homar force-pushed the homar/test_using_trino_filesystem_in_trino_hive_for_orc branch from bafffe5 to 8f35ca7 Compare October 25, 2022 21:24

sopel39 reviewed Oct 26, 2022

View reviewed changes

alexjo2144 reviewed Oct 26, 2022

View reviewed changes

homar force-pushed the homar/test_using_trino_filesystem_in_trino_hive_for_orc branch 2 times, most recently from c15232f to 60d4f1b Compare October 27, 2022 08:29

homar requested review from sopel39 and alexjo2144 October 27, 2022 19:17

homar force-pushed the homar/test_using_trino_filesystem_in_trino_hive_for_orc branch from 60d4f1b to 6e815fa Compare October 27, 2022 20:31

sopel39 approved these changes Oct 28, 2022

View reviewed changes

Port OrcFileWriterFactory.createOrcDataSink to TrinoFileSystem

1782f52

homar force-pushed the homar/test_using_trino_filesystem_in_trino_hive_for_orc branch from 6e815fa to 1782f52 Compare October 28, 2022 11:03

empty

1dff4b8

homar requested a review from sopel39 October 28, 2022 23:37

sopel39 reviewed Nov 2, 2022

View reviewed changes

sopel39 merged commit 83432fe into trinodb:master Nov 2, 2022

github-actions bot added this to the 402 milestone Nov 2, 2022

colebow mentioned this pull request Nov 2, 2022

Add Trino 402 release notes #14862

Merged

homar mentioned this pull request Nov 2, 2022

Port OrcPageSourceFactory completely to TrinoFileSystem #14873

Merged


		Optional<Supplier<OrcDataSource>> validationInputFactory = Optional.empty();
		String stringPath = path.toString();

		@@ -636,22 +638,34 @@ public HiveWriter createWriter(Page partitionColumns, int position, OptionalInt
		hiveWriterStats);
		}

		String file = tempFile.getPath();
		fileSystem.deleteFile(file);

	String file = tempFile.getPath();
	fileSystem.deleteFile(file);
	fileSystem.deleteFile(tempFile.getPath());

Port OrcFileWriterFactory.createOrcDataSink to TrinoFileSystem #14627

Port OrcFileWriterFactory.createOrcDataSink to TrinoFileSystem #14627

Conversation

homar commented Oct 13, 2022 • edited Loading

Description

Non-technical explanation

Release notes

homar commented Oct 14, 2022

sopel39 commented Oct 14, 2022

homar commented Oct 14, 2022 • edited Loading

alexjo2144 commented Oct 17, 2022

sopel39 commented Oct 17, 2022

homar commented Oct 17, 2022

sopel39 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

homar Oct 24, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

homar commented Oct 24, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sopel39 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sopel39 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

homar commented Oct 28, 2022

homar commented Oct 28, 2022

sopel39 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

homar commented Oct 13, 2022 •

edited

Loading

homar commented Oct 14, 2022 •

edited

Loading

homar Oct 24, 2022 •

edited

Loading