feat: add parquet reader

docs/advanced_guidance/json_schemas/contract/components/readable_entity.schema.json

@@ -28,6 +28,11 @@
                             "$ref": "#/$defs/xmlReaderConfig",
                             "description": "Configuration for loading an XML file.",
                             "required": ["reader"]
+                        },
+                        ".parquet": {
+                            "$ref": "#/$defs/parquetReaderConfig",
+                            "description": "Configuration for loading a Parquet file.",
+                            "required": ["reader"]
                         }
                     },
                     "minProperties": 1
@@ -150,6 +155,32 @@
                     }
                 }
             ]
+        },
+        "parquetReaderConfig": {
+            "oneOf": [
+                {
+                    "type": "object",
+                    "properties": {
+                        "reader": {
+                            "const": "DuckDBParquetReader"
+                        },
+                        "kwargs": {
+                            "$ref": "reader_constraints/ddb_parquet_reader.schema.json"
+                        }
+                    }
+                },
+                {
+                    "type": "object",
+                    "properties": {
+                        "reader": {
+                            "const": "SparkParquetReader"
+                        },
+                        "kwargs": {
+                            "$ref": "reader_constraints/spark_parquet_reader.schema.json"
+                        }
+                    }
+                }
+            ]
         }
     },
     "required": ["fields", "reader_config"]

docs/advanced_guidance/json_schemas/contract/components/reader_constraints/ddb_parquet_reader.schema.json

@@ -0,0 +1,12 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "data-ingest:contract/components/reader_constraints/ddb_parquet_reader.schema.json",
+    "title": "Keyword Arguments a DuckDBParquetReader",
+    "description": "Arguments to contol how the DuckDB Parquet Reader interacts with a parquet files",
+    "type": "object",
+    "anyOf": [
+        {
+            "$ref": "global_parquet_reader_args.schema.json"
+        }
+    ]
+}

docs/advanced_guidance/json_schemas/contract/components/reader_constraints/global_parquet_reader_args.schema.json

@@ -0,0 +1,21 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "data-ingest:contract/components/reader_constraints/global_parquet_reader_args.schema.json",
+    "title": "Keyword Arguments used across all Parquet readers",
+    "description": "Arguments present in all Parquet readers available.",
+    "type": "object",
+    "properties": {
+        "hive_partitioning": {
+            "type": "boolean",
+            "description": "Infer statistics and schema from Hive partitioned URL and use them to prune reads."
+        },
+        "field_check_error_code": {
+            "type": "string",
+            "description": "Error code to raise when fields are missing or unexpected."
+        },
+        "field_check_error_message": {
+            "type": "string",
+            "description": "Error message to raise when fields are missing or unexpected."
+        }
+    }
+}

docs/advanced_guidance/json_schemas/contract/components/reader_constraints/spark_parquet_reader.schema.json

@@ -0,0 +1,32 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "data-ingest:contract/components/reader_constraints/spark_parquet_reader.schema.json",
+    "title": "Keyword Arguments a SparkParquetReader",
+    "description": "Arguments to contol how the PySpark Parquet Reader interacts with a parquet files",
+    "type": "object",
+    "anyOf": [
+        {
+            "$ref": "global_parquet_reader_args.schema.json"
+        }
+    ],
+    "properties": {
+        "datetime_rebase_mode": {
+            "type": "string",
+            "description": "The datetimeRebaseMode option allows to specify the rebasing mode for the values of the DATE, TIMESTAMP_MILLIS, TIMESTAMP_MICROS logical types from the Julian to Proleptic Gregorian calendar. Default is `\"EXCEPTION\"`",
+            "enum": [
+                "EXCEPTION",
+                "CORRECTED",
+                "LEGACY"
+            ]
+        },
+        "int96_rebase_mode": {
+            "type": "string",
+            "description": "The int96RebaseMode option allows to specify the rebasing mode for INT96 timestamps from the Julian to Proleptic Gregorian calendar. Default is `\"EXCEPTION\"`.",
+            "enum": [
+                "EXCEPTION",
+                "CORRECTED",
+                "LEGACY"
+            ]
+        }
+    }
+}

docs/advanced_guidance/package_documentation/readers.md

@@ -54,6 +54,25 @@
             members:
                 - __init__
 
+## Parquet
+
+=== "DuckDB"
+
+    ::: dve.core_engine.backends.implementations.duckdb.readers.parquet.DuckDBParquetReader
+        options:
+            heading_level: 3
+            members:
+                - __init__
+
+=== "Spark"
+
+    ::: dve.core_engine.backends.implementations.spark.readers.parquet.SparkParquetReader
+        options:
+            heading_level: 3
+            members:
+                - __init__
+
+
 ## XML
 
 === "Base"

docs/user_guidance/file_transformation.md

@@ -159,8 +159,9 @@ The secondary use of the File Transformation stage is the ability to normalise y
 
 ## Supported Formats
 
-| Format  | DuckDB             | Spark              | Version Available |
-| ------- | ------------------ | ------------------ | ----------------- |
-| `.csv`  | :white_check_mark: | :white_check_mark: | >= 0.1.0 |
-| `.json` | :white_check_mark: | :white_check_mark: | >= 0.1.0 |
-| `.xml`  | :white_check_mark: | :white_check_mark: | >= 0.1.0 |
+| Format     | DuckDB             | Spark              | Version Available |
+| ---------- | ------------------ | ------------------ | ----------------- |
+| `.csv`     | :white_check_mark: | :white_check_mark: | >= 0.1.0          |
+| `.json`    | :white_check_mark: | :white_check_mark: | >= 0.1.0          |
+| `.xml`     | :white_check_mark: | :white_check_mark: | >= 0.1.0          |
+| `.parquet` | :white_check_mark: | :white_check_mark: | >= 0.10.0         |

pyproject.toml

@@ -125,6 +125,10 @@ module = "polars.*"
 follow_imports = "skip"
 # ^language server knows what's going on, but mypy can't find attributes on Self? type
 
+[[tool.mypy.overrides]]
+module = "pyarrow.*"
+ignore_missing_imports = true
+
 [tool.black]
 line-length = 100
 

scripts/testdata_gen/gen_flights_parquet_testdata.py

@@ -0,0 +1,39 @@
+import polars as pl
+from pathlib import Path
+
+
+def main():
+    data = [
+        {
+            "flight_id": 1,
+            "plane_id": 1,
+            "passengers": [
+                {"flight_id": 1, "passenger_id": 1, "passenger_name": "Alice"},
+                {"flight_id": 1, "passenger_id": 2, "passenger_name": "Bob"}
+            ]
+        },
+        {
+            "flight_id": 2,
+            "plane_id": 1,
+            "passengers": [
+                {"flight_id": 2, "passenger_id": 3, "passenger_name": "Charlie"},
+                {"flight_id": 2, "passenger_id": 4, "passenger_name": "Diana"}
+            ]
+        },
+        {
+            "flight_id": 3,
+            "plane_id": 2,
+            "passengers": [
+                {"flight_id": 3, "passenger_id": 5, "passenger_name": "Eve"},
+                {"flight_id": 3, "passenger_id": 6, "passenger_name": "Frank"}
+            ]
+        }
+    ]
+
+    df = pl.DataFrame(data)
+    df.write_parquet(
+        Path(Path(__file__).parent.parent.parent, "tests", "testdata", "flights", "flights.parquet")
+    )
+
+if __name__ == "__main__":
+    main()

src/dve/core_engine/backends/base/reader.py

@@ -5,6 +5,7 @@
 from inspect import ismethod
 from typing import Any, ClassVar, Optional, TypeVar
 
+import pyarrow.parquet as pq
 from pydantic import BaseModel
 from typing_extensions import Protocol
 
@@ -172,8 +173,27 @@ def _check_likely_text_file(resource: URI) -> bool:
         return True
 
     def raise_if_not_sensible_file(self, resource: URI, entity_name: str):
-        """Sense check that the file is a text file. Raise error if doesn't
+        """Sense check that the file is a text file or a valid parquet file. Raise error if doesn't
         appear to be the case."""
+        if resource.endswith(".parquet"):
+            try:
+                pq.ParquetFile(resource)
+            except Exception as exc:
+                raise MessageBearingError(
+                    "The submitted file doesn't appear to be a valid parquet format",
+                    messages=[
+                        FeedbackMessage(
+                            entity=entity_name,
+                            record=None,
+                            failure_type="submission",
+                            error_location="Whole File",
+                            error_code="MalformedFile",
+                            error_message="The resource doesn't seem to be a valid parquet file."
+                        )
+                    ]
+                ) from exc
+            return
+
         if not self._check_likely_text_file(resource):
             raise MessageBearingError(
                 "The submitted file doesn't appear to be text",

src/dve/core_engine/backends/exceptions.py

@@ -57,6 +57,30 @@ def __init__(
         )
 
 
+class UnableToParseParquetError(MessageBearingError):
+    """An error raised when unable to parse a CSV file"""
+
+    def __init__(
+        self, entity_name: str, field_check_error_message: str, field_check_error_code: str
+    ):
+        super().__init__(
+            messages=[
+                FeedbackMessage(
+                    entity="parquet_structure",
+                    record={
+                        entity_name: "Unable to parse file. Please check the structure of the file."
+                    },
+                    failure_type="submission",
+                    is_informational=False,
+                    error_type="parquet read",
+                    error_location=entity_name,
+                    error_message=field_check_error_message,
+                    error_code=field_check_error_code,
+                )
+            ]
+        )
+
+
 class BackendErrorMixin(ABC, BackendError):
     """A mixin used to create backend error type."""
 

src/dve/core_engine/backends/implementations/duckdb/__init__.py

@@ -7,6 +7,7 @@
 from .readers import (
     DuckDBCSVReader,
     DuckDBCSVRepeatingHeaderReader,
+    DuckDBParquetReader,
     DuckDBXMLStreamReader,
     PolarsToDuckDBCSVReader,
 )
@@ -16,6 +17,7 @@
 register_reader(DuckDBCSVReader)
 register_reader(DuckDBCSVRepeatingHeaderReader)
 register_reader(DuckDBJSONReader)
+register_reader(DuckDBParquetReader)
 register_reader(DuckDBXMLStreamReader)
 register_reader(PolarsToDuckDBCSVReader)
 

src/dve/core_engine/backends/implementations/duckdb/contract.py

@@ -30,15 +30,15 @@
     duckdb_read_parquet,
     duckdb_record_index,
     duckdb_write_parquet,
-    get_duckdb_cast_statement_from_annotation,
+    generate_duckdb_casting_statements_from_model,
     get_duckdb_type_from_annotation,
     relation_is_empty,
 )
 from dve.core_engine.backends.implementations.duckdb.types import DuckDBEntities
 from dve.core_engine.backends.metadata.contract import DataContractMetadata
 from dve.core_engine.backends.types import StageSuccessful
 from dve.core_engine.backends.utilities import get_polars_type_from_annotation, stringify_model
-from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME
+from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME, SKIP_CONTRACT_CASTING
 from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import URI, EntityLocations
 from dve.core_engine.validation import RowValidator, apply_row_validator_helper
@@ -165,29 +165,26 @@
                 if RECORD_INDEX_COLUMN_NAME not in relation.columns:
                     relation = self.add_record_index(relation)
 
-                casting_statements = [
-                    (
-                        get_duckdb_cast_statement_from_annotation(column, mdl_fld.annotation)
-                        + f""" AS "{column}" """
-                        if column in relation.columns
-                        else f"CAST(NULL AS {ddb_schema[column]}) AS {column}"
+                if list(
+                    contract_metadata.reader_metadata[entity_name].keys()
+                )[0] not in SKIP_CONTRACT_CASTING:
+                    casting_statements = generate_duckdb_casting_statements_from_model(
+                        model_fields=entity_fields,
+                        rel=relation,
+                        ddb_schema=ddb_schema,
+                        row_index_present=True,
                     )
-                    for column, mdl_fld in entity_fields.items()
-                ]
-                casting_statements.append(
-                    f"CAST({RECORD_INDEX_COLUMN_NAME} AS {get_duckdb_type_from_annotation(int)}) AS {RECORD_INDEX_COLUMN_NAME}"  # pylint: disable=C0301
-                )
-                try:
-                    relation = relation.project(", ".join(casting_statements))
-                except Exception as err:  # pylint: disable=broad-except
-                    successful = False
-                    self.logger.error(f"Error in casting relation: {err}")
-                    dump_processing_errors(
-                        working_dir,
-                        "data_contract",
-                        [generate_error_casting_entity_message(entity_name)],
-                    )
-                    continue
+                    try:
+                        relation = relation.project(", ".join(casting_statements))
+                    except Exception as err:  # pylint: disable=broad-except
+                        successful = False
+                        self.logger.error(f"Error in casting relation: {err}")
+                        dump_processing_errors(
+                            working_dir,
+                            "data_contract",
+                            [generate_error_casting_entity_message(entity_name)],
+                        )
+                        continue
 
                 if self.debug:
                     # count will force evaluation - only done in debug

src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py

@@ -17,6 +17,7 @@
 from duckdb.typing import DuckDBPyType
 from pandas import DataFrame
 from pydantic import BaseModel
+from pydantic.fields import FieldInfo
 from typing_extensions import Annotated, get_args, get_origin, get_type_hints
 
 from dve.common.error_utils import get_feedback_errors_uri
@@ -484,3 +485,27 @@ def get_duckdb_cast_statement_from_annotation(
                 stmt = f"TRIM({quoted_name})"
                 return _cast_as_ddb_type(stmt, type_) if parent_element else stmt
     raise ValueError(f"No equivalent DuckDB type for {type_annotation!r}")
+
+
+def generate_duckdb_casting_statements_from_model(
+    model_fields: dict[str, FieldInfo],
+    rel: DuckDBPyRelation,
+    ddb_schema: dict[str, Any],
+    row_index_present: bool = False,
+) -> list[str]:
+    """Generate duckdb casting statement from pydantic model fields"""
+    casting_statements = [
+        (
+            get_duckdb_cast_statement_from_annotation(column, mdl_fld.annotation)
+            + f""" AS "{column}" """
+            if column in rel.columns
+            else f"CAST(NULL AS {ddb_schema[column]}) AS {column}"
+        )
+        for column, mdl_fld in model_fields.items()
+    ]
+    if row_index_present:
+        casting_statements.append(
+            f"CAST({RECORD_INDEX_COLUMN_NAME} AS {get_duckdb_type_from_annotation(int)}) AS {RECORD_INDEX_COLUMN_NAME}"  # pylint: disable=C0301
+        )
+
+    return casting_statements