diff --git a/CHANGELOG.md b/CHANGELOG.md
index e371d41..5987d64 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,8 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- Parquet read/write is now backed by **DuckDB.jl** instead of **Parquet2.jl**. The
+  previous Parquet2-written files were not reliably readable by all Parquet consumers
+  (e.g. DuckDB, pandas/pyarrow); DuckDB-written output is broadly compatible. `Parquet2`
+  has been dropped as a dependency (replaced by `DuckDB` and `DBInterface`). Both
+  `dbn_to_parquet` and `parquet_to_dbn` are affected. `dbn_to_parquet` writes a single
+  Parquet file at the exact `output_file` path (pass a `*.parquet` path, not a directory).
+
 ### Added
 
+- `dbn_to_parquet` now compresses with **ZSTD by default** and accepts a `compression`
+  keyword — one of `"zstd"` (default), `"snappy"`, `"gzip"`, or `"uncompressed"`.
 - Symbol resolution helpers that join the human-readable `raw_symbol` back onto
   records using `Metadata.mappings`: `symbol_map(metadata)` builds an
   `instrument_id -> [(start_date, end_date, raw_symbol)]` lookup,
@@ -22,6 +33,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- `records_to_dataframe` (and therefore `dbn_to_csv` / `dbn_to_parquet`) no longer throws
+  `FieldError` on **OHLCV**, **STATUS**, **IMBALANCE**, and **DEFINITION** records. The
+  converters referenced struct fields that do not exist: `OHLCVMsg.ts_recv`;
+  `StatusMsg.ts_in_delta`/`sequence` (and a bogus `Action.T` cast of the status action
+  code); `ImbalanceMsg.auction_price`/`ts_in_delta`/`sequence`; and
+  `InstrumentDefMsg.multiplier`/`min_price_increment_portfolio_type`/`ts_in_delta`/
+  `sequence`. They now read the real fields (e.g. STATUS emits `reason`,
+  `trading_event`, `is_trading`/`is_quoting`/`is_short_sell_restricted`; IMBALANCE emits
+  `auction_time`, `ssr_filling_price`, `ind_match_price`, `upper_collar`, `lower_collar`;
+  DEFINITION emits `tick_rule`).
+- `dbn_to_parquet` no longer errors on an **empty record set** (e.g. an OHLCV-1d window
+  with no bars). The exported Parquet preserves the file's real schema (built from
+  `Metadata.schema` via `empty_dataframe_for_schema`) as a valid, zero-row file that reads
+  back as zero records, instead of raising a DuckDB "Table function must return at least
+  one column" error. `mbp10_to_dataframe` likewise returns a typed zero-row frame on empty
+  input rather than a column-less one.
+- `create_metadata_from_dataframe` (used by `parquet_to_dbn` / `csv_to_dbn`) no longer
+  throws from `minimum`/`maximum` on a zero-row DataFrame; an empty frame yields
+  `start_ts = end_ts = 0`, so a fully-empty Parquet/CSV round-trips back to DBN.
 - `records_to_dataframe` (and therefore `to_dataframe` / `dbn_to_parquet`) no
   longer throws `FieldError` on **TBBO / MBP-1** and **MBP-10**. The converters
   read non-existent flat `bid_px_00…` fields; the bid/ask data actually lives in
diff --git a/Project.toml b/Project.toml
index 5cd676e..a29a645 100644
--- a/Project.toml
+++ b/Project.toml
@@ -1,33 +1,35 @@
 name = "DatabentoBinaryEncoding"
 uuid = "90689371-c8cb-40d1-831f-18033db90f74"
+version = "0.1.5"
 authors = ["Tyler Beason <tbeas12@gmail.com>"]
-version = "0.1.4"
 
 [deps]
 CSV = "336ed68f-0bac-5ca0-87d4-7b16caf5d00b"
 CodecZstd = "6b39b394-51ab-5f42-8807-6242bab2b4c2"
+DBInterface = "a10d1c49-ce27-4219-8d33-6db1a4562965"
 DataFrames = "a93c6f00-e57d-5684-b7b6-d8193f3e46c0"
 Dates = "ade2ca70-3891-5945-98fb-dc099432e06a"
+DuckDB = "d2f5444f-75bc-4fdf-ac35-56f514c445e1"
 EnumX = "4e289a0a-7415-4d19-859d-a7e5c4648b56"
 JSON3 = "0f8b85d8-7281-11e9-16c2-39a750bddbf1"
-Parquet2 = "98572fba-bba0-415d-956f-fa77e587d26d"
 Statistics = "10745b16-79ce-11e8-11f9-7d13ad32a3b2"
 StructTypes = "856f2bd8-1eba-4b0a-8007-ebc267875bd4"
 TranscodingStreams = "3bb67fe8-82b1-5028-8e26-92a6c54297fa"
 
 [compat]
-julia = "1.12"
 BenchmarkTools = "1.6.0"
 CSV = "0.10.15"
 CodecZstd = "0.8.6"
+DBInterface = "2.6.1"
 DataFrames = "1.7.0"
 Dates = "1.11.0"
+DuckDB = "1.5.2"
 EnumX = "1.0.5"
 JSON3 = "1.14.3"
-Parquet2 = "0.2.31"
 Statistics = "1.11.0"
 StructTypes = "1.11.0"
 TranscodingStreams = "0.11.3"
+julia = "1.12"
 
 [extras]
 BenchmarkTools = "6e4b80f9-dd63-53aa-95a3-0cdb28fa8baf"
@@ -35,5 +37,5 @@ Documenter = "e30172f5-a6a5-5a46-863b-614d45cd2de4"
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
 
 [targets]
-test = ["Test", "BenchmarkTools"]
 docs = ["Documenter"]
+test = ["Test", "BenchmarkTools"]
diff --git a/README.md b/README.md
index c5f80a3..3228f87 100644
--- a/README.md
+++ b/README.md
@@ -149,7 +149,7 @@ close_writer!(writer)
 # Convert to different formats
 dbn_to_csv("trades.dbn", "trades.csv")
 dbn_to_json("trades.dbn", "trades.json")
-dbn_to_parquet("trades.dbn", "output_dir/")
+dbn_to_parquet("trades.dbn", "trades.parquet")  # ZSTD-compressed by default
 
 # Convert to DataFrame for analysis
 df = records_to_dataframe(records)
diff --git a/docs/src/api/conversion.md b/docs/src/api/conversion.md
index 844c3fb..361f776 100644
--- a/docs/src/api/conversion.md
+++ b/docs/src/api/conversion.md
@@ -41,8 +41,8 @@ dbn_to_csv("trades.dbn", "trades.csv")
 # Convert to JSON
 dbn_to_json("trades.dbn", "trades.json")
 
-# Convert to Parquet
-dbn_to_parquet("trades.dbn", "output_dir/")
+# Convert to Parquet (ZSTD-compressed by default)
+dbn_to_parquet("trades.dbn", "trades.parquet")
 
 # Convert to DataFrame for analysis
 records = read_trades("trades.dbn")
diff --git a/docs/src/guide/conversion.md b/docs/src/guide/conversion.md
index 9f92d71..b0a5b3e 100644
--- a/docs/src/guide/conversion.md
+++ b/docs/src/guide/conversion.md
@@ -86,17 +86,18 @@ dbn_to_json("trades.dbn", "trades.jsonl")
 ```julia
 using DatabentoBinaryEncoding
 
-# Convert to Parquet
-dbn_to_parquet("trades.dbn", "output_directory/")
+# Convert to Parquet (ZSTD-compressed by default)
+dbn_to_parquet("trades.dbn", "trades.parquet")
 
-# Output creates trades.parquet in the directory
+# Choose a different compression codec
+dbn_to_parquet("trades.dbn", "trades.parquet", compression="snappy")
 ```
 
 **Parquet Output:**
 - Columnar format (efficient for analytics)
 - Preserves data types
-- Automatic compression
-- Compatible with Arrow, DuckDB, pandas, etc.
+- ZSTD compression by default (`compression` keyword: `"zstd"`, `"snappy"`, `"gzip"`, `"uncompressed"`)
+- Written via DuckDB, so it is reliably readable by Arrow, DuckDB, pandas/pyarrow, etc.
 
 ### DBN to DataFrame
 
@@ -274,7 +275,7 @@ rm("latest.dbn")
 using DatabentoBinaryEncoding
 
 # Convert historical data to Parquet
-dbn_to_parquet("2024_trades.dbn.zst", "data_lake/")
+dbn_to_parquet("2024_trades.dbn.zst", "data_lake/trades.parquet")
 
 # Now you can query with DuckDB, pandas, etc.
 # SELECT AVG(price) FROM 'data_lake/trades.parquet'
diff --git a/docs/src/index.md b/docs/src/index.md
index 1adabf0..57e499b 100644
--- a/docs/src/index.md
+++ b/docs/src/index.md
@@ -43,7 +43,7 @@ end
 
 # Convert to other formats
 dbn_to_csv("trades.dbn", "trades.csv")
-dbn_to_parquet("trades.dbn", "output_dir/")
+dbn_to_parquet("trades.dbn", "trades.parquet")
 ```
 
 ## Performance Characteristics
diff --git a/docs/src/installation.md b/docs/src/installation.md
index 0cd5ca2..a8b410e 100644
--- a/docs/src/installation.md
+++ b/docs/src/installation.md
@@ -56,7 +56,7 @@ DatabentoBinaryEncoding.jl has the following dependencies (automatically install
 - **CodecZstd** - Zstd compression support
 - **CSV** - CSV file conversion
 - **JSON3** - JSON file conversion
-- **Parquet2** - Parquet file conversion
+- **DuckDB** - Parquet file conversion (read/write)
 - **DataFrames** - DataFrame conversion
 - **EnumX** - Enhanced enum support
 - **StructTypes** - Type serialization
diff --git a/docs/src/quickstart.md b/docs/src/quickstart.md
index 5ba4d93..41bcbee 100644
--- a/docs/src/quickstart.md
+++ b/docs/src/quickstart.md
@@ -153,8 +153,8 @@ dbn_to_csv("trades.dbn", "trades.csv")
 # DBN → JSON
 dbn_to_json("trades.dbn", "trades.json")
 
-# DBN → Parquet
-dbn_to_parquet("trades.dbn", "output_dir/")
+# DBN → Parquet (ZSTD-compressed by default)
+dbn_to_parquet("trades.dbn", "trades.parquet")
 
 # DBN → DataFrame
 df = records_to_dataframe(records)
diff --git a/docs/src/troubleshooting.md b/docs/src/troubleshooting.md
index b066028..8c9c219 100644
--- a/docs/src/troubleshooting.md
+++ b/docs/src/troubleshooting.md
@@ -352,8 +352,10 @@ end
 **Solution**: Ensure Parquet has correct schema:
 ```julia
 # Check column names
-using Parquet2
-df = Parquet2.Dataset("file.parquet") |> DataFrame
+using DuckDB, DBInterface, DataFrames
+con = DBInterface.connect(DuckDB.DB)
+df = DBInterface.execute(con, "SELECT * FROM read_parquet('file.parquet')") |> DataFrame
+DBInterface.close!(con)
 names(df)  # Should match DBN field names
 
 # Ensure correct types (especially timestamps, prices)
diff --git a/src/DatabentoBinaryEncoding.jl b/src/DatabentoBinaryEncoding.jl
index df20e3d..36d50d3 100644
--- a/src/DatabentoBinaryEncoding.jl
+++ b/src/DatabentoBinaryEncoding.jl
@@ -85,7 +85,8 @@ using TranscodingStreams
 using EnumX
 using DataFrames
 using CSV
-using Parquet2
+using DuckDB
+using DBInterface
 using JSON3
 using StructTypes
 
diff --git a/src/export.jl b/src/export.jl
index 1f66ef5..a135f5b 100644
--- a/src/export.jl
+++ b/src/export.jl
@@ -4,9 +4,51 @@ Export functionality for converting DBN data to other formats.
 
 using CSV
 using JSON3
-using Parquet2
+using DuckDB
+using DBInterface
 using DataFrames
 
+# --- Parquet I/O backend (DuckDB) ------------------------------------------
+# Parquet read/write is backed by DuckDB so the output is reliably readable by
+# every Parquet consumer (DuckDB, Arrow, pandas/pyarrow, ...). These private
+# helpers own the connection lifecycle and SQL path handling for both the
+# export (`dbn_to_parquet`) and import (`parquet_to_dbn`) directions.
+
+# Escape a path for a DuckDB SQL string literal: double single quotes, and on
+# Windows use forward slashes (accepted by DuckDB, avoids backslash ambiguity).
+_duckdb_sql_path(p::AbstractString) = replace(replace(p, '\\' => '/'), "'" => "''")
+
+function _write_parquet(df::DataFrame, output_file::AbstractString; compression="zstd")
+    comp = uppercase(String(compression))   # ZSTD | SNAPPY | GZIP | UNCOMPRESSED
+    # DuckDB cannot COPY a zero-column frame, which is what `records_to_dataframe`
+    # returns for an empty record set (e.g. a schema with no rows in the window).
+    # Fall back to a minimal header-only schema so we still write a valid, empty
+    # Parquet file that reads back as zero records.
+    if ncol(df) == 0
+        df = DataFrame(ts_event = Int64[], instrument_id = UInt32[], publisher_id = UInt16[])
+    end
+    con = DBInterface.connect(DuckDB.DB)
+    try
+        DuckDB.register_data_frame(con, df, "_dbn_export")
+        DBInterface.execute(con,
+            "COPY _dbn_export TO '$(_duckdb_sql_path(output_file))' " *
+            "(FORMAT PARQUET, COMPRESSION $comp)")
+    finally
+        DBInterface.close!(con)
+    end
+    return output_file
+end
+
+function _read_parquet(input_file::AbstractString)
+    con = DBInterface.connect(DuckDB.DB)
+    try
+        return DBInterface.execute(con,
+            "SELECT * FROM read_parquet('$(_duckdb_sql_path(input_file))')") |> DataFrame
+    finally
+        DBInterface.close!(con)
+    end
+end
+
 """
     dbn_to_csv(input_file, output_file)
 
@@ -69,23 +111,35 @@ function dbn_to_json(input_file::String, output_file::String; pretty=false)
 end
 
 """
-    dbn_to_parquet(input_file, output_file)
+    dbn_to_parquet(input_file, output_file; compression="zstd")
 
 Convert a DBN file to Parquet format.
 
+The Parquet file is written via DuckDB so the output is reliably readable by every
+Parquet consumer (DuckDB, Arrow, pandas/pyarrow, ...). A single Parquet file is written
+at `output_file`.
+
 # Arguments
 - `input_file::String`: Path to input DBN file
 - `output_file::String`: Path to output Parquet file
+- `compression`: Parquet compression codec — one of `"zstd"` (default), `"snappy"`,
+  `"gzip"`, or `"uncompressed"`.
 
 # Example
 ```julia
 dbn_to_parquet("data.dbn", "data.parquet")
+dbn_to_parquet("data.dbn", "data.parquet", compression="snappy")
 ```
 """
-function dbn_to_parquet(input_file::String, output_file::String)
+function dbn_to_parquet(input_file::String, output_file::String; compression="zstd")
     metadata, records = read_dbn_with_metadata(input_file)
     df = records_to_dataframe(records)
-    Parquet2.writefile(output_file, df)
+    # An empty record set yields a column-less DataFrame; rebuild it from the file's
+    # schema so the Parquet keeps the real columns and round-trips to zero records.
+    if ncol(df) == 0
+        df = empty_dataframe_for_schema(metadata.schema)
+    end
+    _write_parquet(df, output_file; compression=compression)
     return df
 end
 
@@ -138,6 +192,55 @@ function records_to_dataframe(records::Vector)
     end
 end
 
+"""
+    empty_dataframe_for_schema(schema)
+
+Build a zero-row DataFrame with the columns of `schema`'s record type.
+
+`records_to_dataframe` returns a column-less `DataFrame()` for an empty record set
+because it has no record to dispatch on. When the originating `Metadata.schema` is known
+(e.g. in `dbn_to_parquet`), use it here instead so the exported file keeps the real schema
+and round-trips back to zero records. Unknown/mixed schemas fall back to a minimal
+header-only frame.
+"""
+function empty_dataframe_for_schema(schema)
+    if schema == Schema.TRADES
+        return trades_to_dataframe(TradeMsg[])
+    elseif schema == Schema.MBO
+        return mbo_to_dataframe(MBOMsg[])
+    elseif schema == Schema.MBP_1 || schema == Schema.TBBO
+        return mbp1_to_dataframe(MBP1Msg[])
+    elseif schema == Schema.CMBP_1
+        return mbp1_to_dataframe(CMBP1Msg[])
+    elseif schema == Schema.CBBO_1S
+        return mbp1_to_dataframe(CBBO1sMsg[])
+    elseif schema == Schema.CBBO_1M
+        return mbp1_to_dataframe(CBBO1mMsg[])
+    elseif schema == Schema.TCBBO
+        return mbp1_to_dataframe(TCBBOMsg[])
+    elseif schema == Schema.BBO_1S
+        return mbp1_to_dataframe(BBO1sMsg[])
+    elseif schema == Schema.BBO_1M
+        return mbp1_to_dataframe(BBO1mMsg[])
+    elseif schema == Schema.MBP_10
+        return mbp10_to_dataframe(MBP10Msg[])
+    elseif schema == Schema.OHLCV_1S || schema == Schema.OHLCV_1M ||
+           schema == Schema.OHLCV_1H || schema == Schema.OHLCV_1D
+        return ohlcv_to_dataframe(OHLCVMsg[])
+    elseif schema == Schema.STATUS
+        return status_to_dataframe(StatusMsg[])
+    elseif schema == Schema.IMBALANCE
+        return imbalance_to_dataframe(ImbalanceMsg[])
+    elseif schema == Schema.STATISTICS
+        return stat_to_dataframe(StatMsg[])
+    elseif schema == Schema.DEFINITION
+        return instrument_def_to_dataframe(InstrumentDefMsg[])
+    else
+        # Unknown / mixed schema: minimal header-only frame.
+        return DataFrame(ts_event = Int64[], instrument_id = UInt32[], publisher_id = UInt16[])
+    end
+end
+
 function trades_to_dataframe(records::Vector{TradeMsg})
     DataFrame(
         ts_event = [r.hd.ts_event for r in records],
@@ -202,6 +305,17 @@ end
 function mbp10_to_dataframe(records::Vector{MBP10Msg})
     # MBP-10 carries 10 price levels in `levels::NTuple{10,BidAskPair}`; expand
     # each record into one row per level (0-indexed in the output).
+    # The row-by-row builder below would collapse to a 0-column frame on empty
+    # input, so return an explicitly-typed zero-row frame in that case.
+    if isempty(records)
+        return DataFrame(
+            ts_event = Int64[], ts_recv = Int64[], instrument_id = UInt32[],
+            publisher_id = UInt16[], level = Int64[], bid_price = Float64[],
+            ask_price = Float64[], bid_size = UInt32[], ask_size = UInt32[],
+            bid_ct = UInt32[], ask_ct = UInt32[], flags = UInt8[],
+            ts_in_delta = Int32[], sequence = UInt32[], action = String[], side = String[]
+        )
+    end
     rows = []
     for record in records
         for (level, pair) in enumerate(record.levels)
@@ -230,9 +344,9 @@ function mbp10_to_dataframe(records::Vector{MBP10Msg})
 end
 
 function ohlcv_to_dataframe(records::Vector{OHLCVMsg})
+    # OHLCVMsg has no ts_recv: it carries only hd, open/high/low/close, and volume.
     DataFrame(
         ts_event = [r.hd.ts_event for r in records],
-        ts_recv = [r.ts_recv for r in records],
         instrument_id = [r.hd.instrument_id for r in records],
         publisher_id = [r.hd.publisher_id for r in records],
         open = [price_to_float(r.open) for r in records],
@@ -244,27 +358,38 @@ function ohlcv_to_dataframe(records::Vector{OHLCVMsg})
 end
 
 function status_to_dataframe(records::Vector{StatusMsg})
+    # StatusMsg has no ts_in_delta/sequence; `action` is a status action code
+    # (UInt16), not a trading Action enum, so emit it raw alongside the flags.
     DataFrame(
         ts_event = [r.hd.ts_event for r in records],
         ts_recv = [r.ts_recv for r in records],
         instrument_id = [r.hd.instrument_id for r in records],
         publisher_id = [r.hd.publisher_id for r in records],
-        ts_in_delta = [r.ts_in_delta for r in records],
-        sequence = [r.sequence for r in records],
-        action = [string(Action.T(r.action)) for r in records]
+        action = [r.action for r in records],
+        reason = [r.reason for r in records],
+        trading_event = [r.trading_event for r in records],
+        is_trading = [r.is_trading for r in records],
+        is_quoting = [r.is_quoting for r in records],
+        is_short_sell_restricted = [r.is_short_sell_restricted for r in records]
     )
 end
 
 function imbalance_to_dataframe(records::Vector{ImbalanceMsg})
+    # ImbalanceMsg has no auction_price / ts_in_delta / sequence fields; it carries
+    # the auction price set below plus auction_time, and side codes are raw bytes.
     DataFrame(
         ts_event = [r.hd.ts_event for r in records],
         ts_recv = [r.ts_recv for r in records],
         instrument_id = [r.hd.instrument_id for r in records],
         publisher_id = [r.hd.publisher_id for r in records],
         ref_price = [price_to_float(r.ref_price) for r in records],
-        auction_price = [price_to_float(r.auction_price) for r in records],
+        auction_time = [r.auction_time for r in records],
         cont_book_clr_price = [price_to_float(r.cont_book_clr_price) for r in records],
         auct_interest_clr_price = [price_to_float(r.auct_interest_clr_price) for r in records],
+        ssr_filling_price = [price_to_float(r.ssr_filling_price) for r in records],
+        ind_match_price = [price_to_float(r.ind_match_price) for r in records],
+        upper_collar = [price_to_float(r.upper_collar) for r in records],
+        lower_collar = [price_to_float(r.lower_collar) for r in records],
         paired_qty = [r.paired_qty for r in records],
         total_imbalance_qty = [r.total_imbalance_qty for r in records],
         market_imbalance_qty = [r.market_imbalance_qty for r in records],
@@ -274,10 +399,8 @@ function imbalance_to_dataframe(records::Vector{ImbalanceMsg})
         auction_status = [r.auction_status for r in records],
         freeze_status = [r.freeze_status for r in records],
         num_extensions = [r.num_extensions for r in records],
-        unpaired_side = [string(r.unpaired_side) for r in records],
-        significant_imbalance = [r.significant_imbalance for r in records],
-        ts_in_delta = [r.ts_in_delta for r in records],
-        sequence = [r.sequence for r in records]
+        unpaired_side = [r.unpaired_side for r in records],
+        significant_imbalance = [r.significant_imbalance for r in records]
     )
 end
 
@@ -314,7 +437,6 @@ function instrument_def_to_dataframe(records::Vector{InstrumentDefMsg})
         currency = [strip_nulls(String(r.currency)) for r in records],
         instrument_class = [string(r.instrument_class) for r in records],
         strike_price = [price_to_float(r.strike_price) for r in records],
-        multiplier = [r.multiplier for r in records],
         expiration = [r.expiration for r in records],
         activation = [r.activation for r in records],
         high_limit_price = [price_to_float(r.high_limit_price) for r in records],
@@ -339,11 +461,9 @@ function instrument_def_to_dataframe(records::Vector{InstrumentDefMsg})
         contract_multiplier = [r.contract_multiplier for r in records],
         contract_multiplier_unit = [r.contract_multiplier_unit for r in records],
         flow_schedule_type = [r.flow_schedule_type for r in records],
-        min_price_increment_portfolio_type = [r.min_price_increment_portfolio_type for r in records],
+        tick_rule = [r.tick_rule for r in records],
         user_defined_instrument = [r.user_defined_instrument for r in records],
-        trading_reference_date = [r.trading_reference_date for r in records],
-        ts_in_delta = [r.ts_in_delta for r in records],
-        sequence = [r.sequence for r in records]
+        trading_reference_date = [r.trading_reference_date for r in records]
     )
 end
 
diff --git a/src/import.jl b/src/import.jl
index c52823b..3dc6fb3 100644
--- a/src/import.jl
+++ b/src/import.jl
@@ -3,7 +3,6 @@ Import functionality for converting other formats to DBN.
 """
 
 using JSON3
-using Parquet2
 using DataFrames
 using CSV
 
@@ -284,9 +283,9 @@ parquet_to_dbn("data.parquet", "data.dbn", schema=Schema.TRADES, dataset="XNAS")
 """
 function parquet_to_dbn(input_file::String, output_file::String; 
                         schema=nothing, dataset="")
-    # Read Parquet file
-    df = Parquet2.readfile(input_file) |> DataFrame
-    
+    # Read Parquet file (via DuckDB)
+    df = _read_parquet(input_file)
+
     # Convert DataFrame to records
     records = dataframe_to_records(df, schema)
     
@@ -689,9 +688,10 @@ function create_metadata_from_dataframe(df::DataFrame, schema, dataset)
         error("No timestamp column found in DataFrame")
     end
     
-    start_ts = minimum(df[!, ts_col])
-    end_ts = maximum(df[!, ts_col])
-    
+    # Guard the empty case: minimum/maximum throw on a zero-row column.
+    start_ts = isempty(df[!, ts_col]) ? Int64(0) : minimum(df[!, ts_col])
+    end_ts = isempty(df[!, ts_col]) ? Int64(0) : maximum(df[!, ts_col])
+
     return Metadata(
         UInt8(3),                    # DBN version
         dataset,                     # dataset