Refactoring HiveConnectorTestBase::writeToFile() to use dynamic cast for writer of DWRF or Parquet file format

[minhancao]

Refactoring HiveConnectorTestBase::writeToFile() to use dynamic cast for writer of DWRF or Parquet file format. Also refactoring HiveIcebergTest and TableScanTest since they use HiveConnectorTestBase class.

Related to work in:
#9973
#10915

Resolves:
#9171

[netlify]

✅ Deploy Preview for meta-velox canceled.

Name	Link
🔨 Latest commit	d1d9dc6
🔍 Latest deploy log	https://app.netlify.com/sites/meta-velox/deploys/670d72a3d9be4200084464ac

[czentgr]

Thanks, a few comments.

[czentgr]

@minhancao please squash your commits and take a look at the failures.

[yingsu00]

Mostly looks good!

[minhancao]

@yingsu00 I have rebased with main and fixed merge conflicts, please review when you can, thank you!

I've noticed that dwrf configs are passed to HiveConnectorTestBase::writeToFile() quite often in tests so I decided to create my own function for agnostic use, HiveConnectorTestBase::writeToFileCommon(). I suggest maybe also changing writeToFile() to be named writeToFileDwrf() since it seems to only pertain to dwrf file format based from its usage in tests. What do you think?

I've also noticed that in velox/dwio/common/Options.h, there is the struct WriterOptions and it has properties that both pertain to dwrf and parquet. It looks like the current approach here is to populate the writer options accordingly based on the WriterFactory's file format. If we go with the current approach, we would be further adding on much more properties to the common struct WriterOptions when more file formats are supported later on... is that ok? Or maybe I can try another route and try having the common WriterOptions as a pointer parameter for HiveConnectorTestBase::writeToFileCommon() and then dynamically casting it to its respective file format?

velox_hive_iceberg_test results:

[----------] Global test environment tear-down
[==========] 6 tests from 2 test suites ran. (44251 ms total)
[  PASSED  ] 6 tests.

velox_hive_connector_test results:

[----------] Global test environment tear-down
[==========] 68 tests from 9 test suites ran. (5355 ms total)
[  PASSED  ] 68 tests.

[yingsu00]

@minhancao Can you please rebase this PR after #10470 and #10505 are merged? Thanks!

[minhancao]

@yingsu00 I have rebased with main to bring in your PR and Pedro's PR, please review when you can, thanks!

I had one error while running the HiveIcebergTest:

[ RUN      ] HiveIcebergTest.positionalDeletesMultipleSplits
I20240724 12:04:58.236274 34091896 PeriodicStatsReporter.cpp:69] Starting PeriodicStatsReporter with options allocatorStatsIntervalMs:2000, cacheStatsIntervalMs:2000, arbitratorStatsIntervalMs:2000, spillStatsIntervalMs:2000
I20240724 12:04:58.236331 34091896 HiveConnector.cpp:72] Hive connector test-hive created with maximum of 20000 cached file handles.
I20240724 12:04:58.236330 34092510 PeriodicStatsReporter.cpp:252] Spill memory usage: current[0B] peak[0B]
I20240724 12:04:58.245956 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_TpBDfq,  total size: 41.54KB
I20240724 12:04:58.252242 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_DKzNVD,  total size: 43.27KB
I20240724 12:04:58.258396 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_xxGkPr,  total size: 43.01KB
I20240724 12:04:58.264377 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_Uj7KCA,  total size: 43.39KB
I20240724 12:04:58.270493 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_QlVa5l,  total size: 42.68KB
I20240724 12:04:58.276564 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_CCIW7Z,  total size: 43.38KB
I20240724 12:04:58.282634 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_I3Z1Ao,  total size: 43.06KB
I20240724 12:04:58.288759 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_QRWNe2,  total size: 43.20KB
I20240724 12:04:58.294911 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_EWrb48,  total size: 43.00KB
I20240724 12:04:58.300951 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_WPeKqV,  total size: 43.01KB
I20240724 12:04:58.337361 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_360vuy,  total size: 823B
I20240724 12:04:58.337754 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_duS6np,  total size: 823B
I20240724 12:04:58.338088 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_BOrWxR,  total size: 824B
I20240724 12:04:58.338434 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_m9vFQ5,  total size: 824B
I20240724 12:04:58.338773 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_c2y9Tc,  total size: 824B
I20240724 12:04:58.339118 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_MWMgIO,  total size: 823B
I20240724 12:04:58.339449 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_RWj82r,  total size: 823B
I20240724 12:04:58.339777 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_PfJ2W7,  total size: 823B
I20240724 12:04:58.340096 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_PXIZzM,  total size: 823B
I20240724 12:04:58.341436 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_bPJjyp,  total size: 824B
I20240724 12:04:58.352560 34092514 Task.cpp:1141] All drivers (1) finished for task test_cursor 65 after running for 10 ms.
I20240724 12:04:58.352576 34092514 Task.cpp:1881] Terminating task test_cursor 65 with state Finished after running for 10 ms.
I20240724 12:05:00.241349 34092510 PeriodicStatsReporter.cpp:252] Spill memory usage: current[0B] peak[0B]
I20240724 12:05:00.602698 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_SOBYCO,  total size: 193.28KB
I20240724 12:05:00.611328 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_QdpiJk,  total size: 193.28KB
I20240724 12:05:00.619948 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_ff5cuf,  total size: 193.28KB
I20240724 12:05:00.628422 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_WTEiyn,  total size: 193.28KB
I20240724 12:05:00.637248 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_Q8k11W,  total size: 193.28KB
I20240724 12:05:00.645998 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_qWU5vX,  total size: 193.28KB
I20240724 12:05:00.654503 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_s7jOsF,  total size: 193.28KB
I20240724 12:05:00.662882 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_hGWN9k,  total size: 193.28KB
I20240724 12:05:00.671310 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_qT6nR9,  total size: 193.28KB
I20240724 12:05:00.679775 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_G9HEaf,  total size: 193.28KB
I20240724 12:05:00.716064 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_gol4yd,  total size: 747B
I20240724 12:05:00.716542 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_Jlp9ZY,  total size: 747B
I20240724 12:05:00.716934 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_x61MYj,  total size: 747B
I20240724 12:05:00.717315 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_Ldaf9L,  total size: 747B
I20240724 12:05:00.717693 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_gxRPjB,  total size: 747B
I20240724 12:05:00.718063 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_VKmSvE,  total size: 747B
I20240724 12:05:00.718425 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_CxXyhB,  total size: 747B
I20240724 12:05:00.718778 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_HPaxTP,  total size: 747B
I20240724 12:05:00.719138 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_F6686O,  total size: 747B
I20240724 12:05:00.719501 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_iFNSsl,  total size: 747B
I20240724 12:05:00.732672 34092610 Task.cpp:1141] All drivers (1) finished for task test_cursor 66 after running for 12 ms.
I20240724 12:05:00.732692 34092610 Task.cpp:1881] Terminating task test_cursor 66 with state Finished after running for 12 ms.
I20240724 12:05:02.246464 34092510 PeriodicStatsReporter.cpp:252] Spill memory usage: current[0B] peak[0B]
I20240724 12:05:02.977259 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_HIXdeQ,  total size: 41.54KB
I20240724 12:05:02.983448 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_Co1iql,  total size: 43.27KB
I20240724 12:05:02.989531 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_3U2Kh3,  total size: 43.01KB
I20240724 12:05:02.995604 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_gizLqI,  total size: 43.39KB
I20240724 12:05:03.001678 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_BcGRuK,  total size: 42.68KB
I20240724 12:05:03.007730 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_zkuKP3,  total size: 43.38KB
I20240724 12:05:03.013906 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_bOn7qm,  total size: 43.06KB
I20240724 12:05:03.019982 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_fQfMxn,  total size: 43.20KB
I20240724 12:05:03.026090 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_O9GIgn,  total size: 43.00KB
I20240724 12:05:03.032234 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_jM1lmu,  total size: 43.01KB
I20240724 12:05:03.068727 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_V3grWJ,  total size: 823B
I20240724 12:05:03.069109 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_N1a1c7,  total size: 823B
I20240724 12:05:03.069439 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_IiiGOy,  total size: 823B
I20240724 12:05:03.069805 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_ZknRAz,  total size: 824B
I20240724 12:05:03.070137 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_8VU7Q1,  total size: 823B
I20240724 12:05:03.070472 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_zgZ17d,  total size: 824B
I20240724 12:05:03.070782 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_ulZppl,  total size: 818B
I20240724 12:05:03.071113 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_LFtiPB,  total size: 823B
I20240724 12:05:03.071430 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_W1Qj8A,  total size: 824B
I20240724 12:05:03.071772 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_HXb7Bi,  total size: 824B
I20240724 12:05:03.084453 34092632 Task.cpp:1141] All drivers (1) finished for task test_cursor 67 after running for 12 ms.
I20240724 12:05:03.084478 34092632 Task.cpp:1881] Terminating task test_cursor 67 with state Finished after running for 12 ms.
I20240724 12:05:04.250772 34092510 PeriodicStatsReporter.cpp:252] Spill memory usage: current[0B] peak[0B]
I20240724 12:05:05.337659 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_mFzrk5,  total size: 193.28KB
I20240724 12:05:05.346154 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_25tZKD,  total size: 193.28KB
I20240724 12:05:05.354758 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_5D8vo9,  total size: 193.28KB
I20240724 12:05:05.363242 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_JXYmWs,  total size: 193.28KB
I20240724 12:05:05.371652 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_Tz6OYQ,  total size: 193.28KB
I20240724 12:05:05.380399 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_6DwCPi,  total size: 193.28KB
I20240724 12:05:05.389014 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_GAXUIJ,  total size: 193.28KB
I20240724 12:05:05.397543 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_eAijvC,  total size: 193.28KB
I20240724 12:05:05.405925 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_WL6R9G,  total size: 193.28KB
I20240724 12:05:05.414340 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_jIOHti,  total size: 193.28KB
I20240724 12:05:05.451403 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_DVAzmV,  total size: 747B
I20240724 12:05:05.451829 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_9jpsY1,  total size: 747B
I20240724 12:05:05.452219 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_AtmLbz,  total size: 747B
I20240724 12:05:05.452610 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_IxfNAL,  total size: 747B
I20240724 12:05:05.452991 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_XmiWIY,  total size: 747B
I20240724 12:05:05.453364 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_JtfXTH,  total size: 747B
I20240724 12:05:05.453716 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_qgsZhB,  total size: 747B
I20240724 12:05:05.454093 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_3HaSAm,  total size: 747B
I20240724 12:05:05.454448 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_w1nUpu,  total size: 747B
I20240724 12:05:05.454800 34091896 FileSink.cpp:125] closing file: /tmp/velox_test_9jgJ1x,  total size: 747B
E20240724 12:05:05.475178 34092720 Exceptions.h:67] Line: /Users/minhancao/velox_2/velox/velox/common/file/File.cpp:118, Function:LocalReadFile, Expression:  open failure in LocalReadFile constructor, -1 /tmp/velox_test_9jpsY1 Too many open files., Source: RUNTIME, ErrorCode: INVALID_STATE
I20240724 12:05:05.475392 34092720 Task.cpp:1881] Terminating task test_cursor 68 with state Failed after running for 20 ms.
I20240724 12:05:05.475462 34092720 Task.cpp:1141] All drivers (1) finished for task test_cursor 68 after running for 20 ms.
unknown file: Failure
C++ exception with description "Exception: VeloxRuntimeError
Error Source: RUNTIME
Error Code: INVALID_STATE
Reason: open failure in LocalReadFile constructor, -1 /tmp/velox_test_9jpsY1 Too many open files.
Retriable: False
Context: Split [Hive: /tmp/velox_test_GAXUIJ 0 - 197922] Task test_cursor 68
Additional Context: Operator: TableScan[0] 0
Function: LocalReadFile
File: /Users/minhancao/velox_2/velox/velox/common/file/File.cpp
Line: 118

Do you maybe have any insight for this error?

Ying's PR:
#10505

Pedro's PRs:
#10470
#10520 (to fix build errors caused by previous PR)

[minhancao]

@yingsu00 @czentgr I noticed from Pedro's recent PR #10470, he moved the parquet specific option properties in dwio/common/Options.h to only be in the parquet::WriterOptions in dwio/parquet/writer/Writer.h. In parquet/writer/Writer.cpp, he also rewrote the createWriter() function to be:

std::unique_ptr<dwio::common::Writer> ParquetWriterFactory::createWriter(
    std::unique_ptr<dwio::common::FileSink> sink,
    const std::shared_ptr<dwio::common::WriterOptions>& options) {
  auto parquetOptions =
      std::dynamic_pointer_cast<parquet::WriterOptions>(options);
  VELOX_CHECK_NOT_NULL(
      parquetOptions,
      "Parquet writer factory expected a Parquet WriterOptions object.");
  return std::make_unique<Writer>(
      std::move(sink), *parquetOptions, asRowType(options->schema));
}

The current route that my PR is taking is providing all options of DWRF and Parquet into dwio/common/Options.h and populating it accordingly based on the file format for its respective writers. I have edited the code above to go to my route when the dynamic cast fails.

std::unique_ptr<dwio::common::Writer> ParquetWriterFactory::createWriter(
    std::unique_ptr<dwio::common::FileSink> sink,
    const std::shared_ptr<dwio::common::WriterOptions>& options) {
  auto parquetOptions =
      std::dynamic_pointer_cast<parquet::WriterOptions>(options);

  if (parquetOptions == nullptr) {
    auto parquetOptionsFromCommon = getParquetOptions(*options);
    return std::make_unique<Writer>(
        std::move(sink), parquetOptionsFromCommon, asRowType(options->schema));
  }

  return std::make_unique<Writer>(
      std::move(sink), *parquetOptions, asRowType(options->schema));
}

Is this ok? Or is there a better way to approach this?

It seems to me that we are in conflict now of whether we should take the route of providing all file format options into dwio/common/Options.h and populating it accordingly OR providing the specific file format WriterOptions at the beginning in the test or wherever it's called and dynamically casting it when it is received to its respective file format type like in Pedro's PR.

[minhancao]

Hi @pedroerp! I saw from your recent PR #10470 you moved the Parquet specific options from dwio/common/Options.h to be in the specific Parquet writer options. Do you think the DWRF options that are in dwio/common/Options.h should be moved to the specific DWRF writer options as well and refactor accordingly?

This PR was meant to refactor HiveConnectorTestBase::writeToFile() to be able to be agnostic and use DWRF or Parquet writer depending on the file format that is passed in.

[pedroerp]

Do you think the DWRF options that are in dwio/common/Options.h should be moved to the specific DWRF writer options as well and refactor accordingly?

Hi @minhancao, yes, for sure. I have a draft PR for that, but it turns out to be a bit more refactoring that I expected, but I'm planning to get back at it next week.

[minhancao]

@pedroerp Ok awesome! Can you please link me the PR whenever it's ready?

[minhancao]

Hi @pedroerp! Just wanted to check in if you were working on a PR for the DWRF options that are in dwio/common/Options.h to be moved to the specific DWRF writer options just like how you did with Parquet options? If not, I can take up this work.

[pedroerp]

Hi @pedroerp! Just wanted to check in if you were working on a PR for the DWRF options that are in dwio/common/Options.h to be moved to the specific DWRF writer options just like how you did with Parquet options? If not, I can take up this work.

@minhancao I haven't had time to look into that yet, but I'm happy to help with code review if this is something you would like to work on. @Yuhta has a lot more context on that code though.

The one tricky part is to make sure we maintain the same behavior and default values; I remember seeing that there a couple of different places where these are defined for DWRF today.

[minhancao]

Looks like DWRF and Parquet writer options were both refactored to be using dynamic_cast now thanks to this PR 2 weeks ago:
#10915

[minhancao]

Refactored HiveConnectorTestBase::writeToFile(), HiveIcebergTest, and TableScanTest based on #10915.

[minhancao]

@czentgr @yingsu00 Please review this PR whenever you can, thanks!

[majetideepak]

CC: @svm1

[stale]

This pull request has been automatically marked as stale because it has not had recent activity. If you'd still like this PR merged, please comment on the PR, make sure you've addressed reviewer comments, and rebase on the latest main. Thank you for your contributions!