Skip to content

ARROW-8062: [C++][Dataset] Implement ParquetDatasetFactory #7180

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

Closed
fsaintjacques wants to merge 7 commits into from
Closed

ARROW-8062: [C++][Dataset] Implement ParquetDatasetFactory #7180

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d332258
ARROW-8062: [C++][Dataset] Implement ParquetDatasetFactory
fsaintjacques Mar 27, 2020
40d27d1
Add RowGroupInfo
fsaintjacques May 18, 2020
3ef3688
add parquet_dataset tests
jorisvandenbossche May 19, 2020
3b85adc
Fix race in Scanner::ToTable
fsaintjacques May 20, 2020
7bd26ee
Add ParquetDatasetFactory support to example
fsaintjacques May 20, 2020
d0a3e3b
Address comments
fsaintjacques May 21, 2020
29f44d9
Fix windows with normalized path
fsaintjacques May 21, 2020
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
60 changes: 52 additions & 8 deletions cpp/examples/arrow/dataset-parquet-scan-example.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,18 +15,18 @@
// specific language governing permissions and limitations
// under the License.

#include <cstdlib>
#include <iostream>

#include <arrow/api.h>

#include <arrow/dataset/dataset.h>
#include <arrow/dataset/discovery.h>
#include <arrow/dataset/file_base.h>
#include <arrow/dataset/file_parquet.h>
#include <arrow/dataset/filter.h>
#include <arrow/dataset/scanner.h>
#include <arrow/filesystem/filesystem.h>
#include <arrow/filesystem/path_util.h>

#include <cstdlib>
#include <iostream>

using arrow::field;
using arrow::int16;
Expand Down Expand Up @@ -73,12 +73,12 @@ std::shared_ptr<fs::FileSystem> GetFileSystemFromUri(const std::string& uri,
return fs::FileSystemFromUri(uri, path).ValueOrDie();
}

std::shared_ptr<ds::Dataset> GetDatasetFromPath(std::shared_ptr<fs::FileSystem> fs,
std::shared_ptr<ds::FileFormat> format,
std::string path) {
std::shared_ptr<ds::Dataset> GetDatasetFromDirectory(
std::shared_ptr<fs::FileSystem> fs, std::shared_ptr<ds::ParquetFileFormat> format,
std::string dir) {
// Find all files under `path`
fs::FileSelector s;
s.base_dir = path;
s.base_dir = dir;
s.recursive = true;

ds::FileSystemFactoryOptions options;
Expand All @@ -97,6 +97,50 @@ std::shared_ptr<ds::Dataset> GetDatasetFromPath(std::shared_ptr<fs::FileSystem>
return dataset.ValueOrDie();
}

std::shared_ptr<ds::Dataset> GetParquetDatasetFromMetadata(
std::shared_ptr<fs::FileSystem> fs, std::shared_ptr<ds::ParquetFileFormat> format,
std::string metadata_path) {
auto factory = ds::ParquetDatasetFactory::Make(metadata_path, fs, format).ValueOrDie();
return factory->Finish().ValueOrDie();
}

std::shared_ptr<ds::Dataset> GetDatasetFromFile(
std::shared_ptr<fs::FileSystem> fs, std::shared_ptr<ds::ParquetFileFormat> format,
std::string file) {
ds::FileSystemFactoryOptions options;
// The factory will try to build a child dataset.
auto factory = ds::FileSystemDatasetFactory::Make(fs, {file}, format, options).ValueOrDie();

// Try to infer a common schema for all files.
auto schema = factory->Inspect(conf.inspect_options).ValueOrDie();
// Caller can optionally decide another schema as long as it is compatible
// with the previous one, e.g. `factory->Finish(compatible_schema)`.
auto child = factory->Finish(conf.finish_options).ValueOrDie();

ds::DatasetVector children{conf.repeat, child};
auto dataset = ds::UnionDataset::Make(std::move(schema), std::move(children));

return dataset.ValueOrDie();
}

std::shared_ptr<ds::Dataset> GetDatasetFromPath(
std::shared_ptr<fs::FileSystem> fs, std::shared_ptr<ds::ParquetFileFormat> format,
std::string path) {
auto info = fs->GetFileInfo(path).ValueOrDie();
if (info.IsDirectory()) {
return GetDatasetFromDirectory(fs, format, path);
}

auto dirname_basename = arrow::fs::internal::GetAbstractPathParent(path);
auto basename = dirname_basename.second;

if (basename == "_metadata") {
return GetParquetDatasetFromMetadata(fs, format, path);
}

return GetDatasetFromFile(fs, format, path);
}

std::shared_ptr<ds::Scanner> GetScannerFromDataset(std::shared_ptr<ds::Dataset> dataset,
std::vector<std::string> columns,
std::shared_ptr<ds::Expression> filter,
Expand Down
Loading