feat: add native Parquet file reading support via extension #71
+2,343
−13
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| # Parquet Extension | ||
|
|
||
| Apache Parquet is a columnar storage format widely used in data engineering and analytics workloads. NeuG supports Parquet file import functionality through the Extension framework. After loading the Parquet Extension, users can directly load external Parquet files using the `LOAD FROM` syntax. | ||
|
|
||
| ## Install Extension | ||
|
|
||
| ```cypher | ||
| INSTALL PARQUET; | ||
| ``` | ||
|
|
||
| ## Load Extension | ||
|
|
||
| ```cypher | ||
| LOAD PARQUET; | ||
| ``` | ||
|
|
||
| ## Using Parquet Extension | ||
|
|
||
| `LOAD FROM` reads Parquet files and exposes their columns for querying. Schema is automatically inferred from the Parquet file metadata by default. | ||
|
|
||
| ### Parquet Format Options | ||
|
|
||
| The following options control how Parquet files are read: | ||
|
|
||
| | Option | Type | Default | Description | | ||
| | ------------------------ | ----- | ------- | ------------------------------------------------------------------------------------------------------------------------------------------- | | ||
| | `buffered_stream` | bool | `true` | Enable buffered I/O stream for improved sequential read performance. | | ||
| | `pre_buffer` | bool | `false` | Pre-buffer column data before decoding. Recommended for high-latency filesystems such as S3. | | ||
| | `enable_io_coalescing` | bool | `true` | Enable Arrow I/O read coalescing (hole-filling cache) to reduce I/O overhead when reading non-contiguous byte ranges. When `true`, uses lazy coalescing; when `false`, uses eager coalescing. | | ||
| | `parquet_batch_rows` | int64 | `65536` | Number of rows per Arrow record batch when converting Parquet row groups into in-memory batches. | | ||
|
|
||
| ### Query Examples | ||
|
|
||
| #### Basic Parquet Loading | ||
|
|
||
| Load all columns from a Parquet file: | ||
|
|
||
| ```cypher | ||
| LOAD FROM "person.parquet" | ||
| RETURN *; | ||
| ``` | ||
|
|
||
| #### Specifying Batch Size | ||
|
|
||
| Tune memory usage by adjusting the number of rows read per batch: | ||
|
|
||
| ```cypher | ||
| LOAD FROM "person.parquet" (parquet_batch_rows=8192) | ||
| RETURN *; | ||
| ``` | ||
|
|
||
| #### Enabling I/O Coalescing | ||
|
|
||
| Enable eager I/O coalescing for workloads that benefit from pre-fetching contiguous data: | ||
|
|
||
| ```cypher | ||
| LOAD FROM "person.parquet" (enable_io_coalescing=false) | ||
| RETURN *; | ||
| ``` | ||
|
|
||
| #### Column Projection | ||
|
|
||
| Return only specific columns from Parquet data: | ||
|
|
||
| ```cypher | ||
| LOAD FROM "person.parquet" | ||
| RETURN fName, age; | ||
| ``` | ||
|
|
||
| #### Column Aliases | ||
|
|
||
| Use `AS` to assign aliases to columns: | ||
|
|
||
| ```cypher | ||
| LOAD FROM "person.parquet" | ||
| RETURN fName AS name, age AS years; | ||
| ``` | ||
|
|
||
| > **Note:** All relational operations supported by `LOAD FROM` — including type conversion, WHERE filtering, aggregation, sorting, and limiting — work the same way with Parquet files. See the [LOAD FROM reference](../data_io/load_data) for the complete list of operations. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,122 @@ | ||
| #!/usr/bin/env python3 | ||
| # Copyright 2020 Alibaba Group Holding Limited. | ||
| # | ||
| # Licensed 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. | ||
|
|
||
| """ | ||
| Preprocess comprehensive_graph CSV files into Parquet format. | ||
|
|
||
| Outputs are written to example_dataset/comprehensive_graph/parquet/. | ||
|
|
||
| Notes: | ||
| - interval_property is kept as a Parquet STRING column (pyarrow CSV inference). | ||
| NeuG's INTERVAL text format ("1year2months3days...") has no writable Parquet | ||
| native interval equivalent via pyarrow, so string is the simplest lossless form. | ||
| - All other columns (INT32/INT64/UINT32/UINT64/FLOAT/DOUBLE/STRING/DATE/DATETIME) | ||
| are preserved with explicitly typed Arrow columns. | ||
| """ | ||
|
|
||
| import os | ||
| import sys | ||
|
|
||
| try: | ||
| import pyarrow as pa | ||
| import pyarrow.csv as pa_csv | ||
| import pyarrow.parquet as pq | ||
| except ImportError: | ||
| print("Error: pyarrow is required. Install with: pip install pyarrow") | ||
| sys.exit(1) | ||
|
|
||
| SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) | ||
| OUTPUT_DIR = os.path.join(SCRIPT_DIR, "parquet") | ||
|
|
||
| # Edge files: all have duplicate src/dst column both named 'node_a.id'. | ||
| # Rename the first two columns to src_id / dst_id for uniqueness in Parquet. | ||
| EDGE_FILES = { | ||
| "rel_a.csv", | ||
| } | ||
|
|
||
| # All CSV files to convert | ||
| CSV_FILES = [ | ||
| "node_a.csv", | ||
| "rel_a.csv", | ||
| ] | ||
|
|
||
|
|
||
| def convert(csv_filename: str) -> None: | ||
| csv_path = os.path.join(SCRIPT_DIR, csv_filename) | ||
| parquet_filename = os.path.splitext(csv_filename)[0] + ".parquet" | ||
| parquet_path = os.path.join(OUTPUT_DIR, parquet_filename) | ||
|
|
||
| read_opts = pa_csv.ReadOptions() | ||
| parse_opts = pa_csv.ParseOptions(delimiter="|") | ||
| # timestamp_parsers: let Arrow auto-detect date/datetime columns. | ||
| # Explicitly type columns where CSV auto-inference picks wrong types: | ||
| # - i32_property: inferred as int64 (pyarrow defaults all integers to int64) | ||
| # - u32/u64_property: inferred as double (values overflow int64) | ||
| # - f32_property: inferred as double (pyarrow has no float32 inference) | ||
| node_column_types = { | ||
| "i32_property": pa.int32(), | ||
| "u32_property": pa.uint32(), | ||
| "u64_property": pa.uint64(), | ||
| "f32_property": pa.float32(), | ||
| } if csv_filename in ("node_a.csv", "node_b.csv") else {} | ||
| edge_column_types = { | ||
| "i32_weight": pa.int32(), | ||
| # datetime_weight values are date-only strings ("2023-05-17"); pyarrow infers | ||
| # date32[day] by default, but DT_DATETIME requires timestamp[ms]. | ||
| "datetime_weight": pa.timestamp("ms"), | ||
| } if csv_filename in EDGE_FILES else {} | ||
| column_types = {**node_column_types, **edge_column_types} | ||
| convert_opts = pa_csv.ConvertOptions( | ||
| timestamp_parsers=["%Y-%m-%dT%H:%M:%S", "%Y-%m-%d"], | ||
| column_types=column_types, | ||
| ) | ||
|
|
||
| table = pa_csv.read_csv( | ||
| csv_path, | ||
| read_options=read_opts, | ||
| parse_options=parse_opts, | ||
| convert_options=convert_opts, | ||
| ) | ||
|
|
||
| # Rename duplicate src/dst columns in edge files | ||
| if csv_filename in EDGE_FILES: | ||
| old_names = table.schema.names | ||
| new_names = ["src_id" if i == 0 else "dst_id" if i == 1 else n | ||
| for i, n in enumerate(old_names)] | ||
| table = table.rename_columns(new_names) | ||
| print(f" renamed src/dst columns: {old_names[:2]} -> ['src_id', 'dst_id']") | ||
|
|
||
| # interval_property is kept as-is: pyarrow CSV infers it as string | ||
| pq.write_table(table, parquet_path) | ||
| print(f" {csv_filename} -> parquet/{parquet_filename} ({table.num_rows} rows, {len(table.schema)} cols)") | ||
|
|
||
|
|
||
| def main(): | ||
| os.makedirs(OUTPUT_DIR, exist_ok=True) | ||
| print(f"Output directory: {OUTPUT_DIR}\n") | ||
|
|
||
| for csv_file in CSV_FILES: | ||
| csv_path = os.path.join(SCRIPT_DIR, csv_file) | ||
| if not os.path.exists(csv_path): | ||
| print(f"[SKIP] {csv_file} not found") | ||
| continue | ||
| print(f"Converting {csv_file} ...") | ||
| convert(csv_file) | ||
|
|
||
| print("\nDone.") | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| main() |
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ARROW_PARQUET itself is a guard in Arrow CMakeList, maybe we should use a different name here, i.e. ARROW_ENABLE_PARQUET