Parquet file doesn't mark date and timestamp columns

[dcory]

The Parquet file metadata doesn't include logical type information that (for instance) EventDate is a date and EventTime is a timestamp.

https://github.com/apache/parquet-format/blob/master/LogicalTypes.md

[alexey-milovidov]

Yes, these Parquet files are just converted from the source data.
It's ok to convert the types during load. Or should we add this information for convenience?

What will be the easiest way to do this?

[dcory]

It would definitely be useful to include this information in the Parquet file so that I don't have to explicitly convert it on load, especially for a stateless table engine.

[gortiz]

Same happen with Strings, which are just binaries that are not being marked as Strings.

Apache Pinot is quite strict about that and when importing data from parquet (which is quite faster than using tsv) the data is not imported as UTF-8 but as a byte array (because that is what parquet says)

For example:

> parq hits_01.parquet --schema

 # Schema 
 <pyarrow._parquet.ParquetSchema object at 0x7f1ccc4c4180>
required group field_id=-1 schema {
  optional int64 field_id=-1 WatchID;
  optional int32 field_id=-1 JavaEnable (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 Title;
  optional int32 field_id=-1 GoodEvent (Int(bitWidth=16, isSigned=true));
  optional int64 field_id=-1 EventTime;
  optional int32 field_id=-1 EventDate (Int(bitWidth=16, isSigned=false));
  optional int32 field_id=-1 CounterID;
  optional int32 field_id=-1 ClientIP;
  optional int32 field_id=-1 RegionID;
  optional int64 field_id=-1 UserID;
  optional int32 field_id=-1 CounterClass (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 OS (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 UserAgent (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 URL;
  optional binary field_id=-1 Referer;
  optional int32 field_id=-1 IsRefresh (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 RefererCategoryID (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 RefererRegionID;
  optional int32 field_id=-1 URLCategoryID (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 URLRegionID;
  optional int32 field_id=-1 ResolutionWidth (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 ResolutionHeight (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 ResolutionDepth (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 FlashMajor (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 FlashMinor (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 FlashMinor2;
  optional int32 field_id=-1 NetMajor (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 NetMinor (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 UserAgentMajor (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 UserAgentMinor;
  optional int32 field_id=-1 CookieEnable (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 JavascriptEnable (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 IsMobile (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 MobilePhone (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 MobilePhoneModel;
  optional binary field_id=-1 Params;
  optional int32 field_id=-1 IPNetworkID;
  optional int32 field_id=-1 TraficSourceID (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 SearchEngineID (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 SearchPhrase;
  optional int32 field_id=-1 AdvEngineID (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 IsArtifical (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 WindowClientWidth (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 WindowClientHeight (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 ClientTimeZone (Int(bitWidth=16, isSigned=true));
  optional int64 field_id=-1 ClientEventTime;
  optional int32 field_id=-1 SilverlightVersion1 (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 SilverlightVersion2 (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 SilverlightVersion3;
  optional int32 field_id=-1 SilverlightVersion4 (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 PageCharset;
  optional int32 field_id=-1 CodeVersion;
  optional int32 field_id=-1 IsLink (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 IsDownload (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 IsNotBounce (Int(bitWidth=16, isSigned=true));
  optional int64 field_id=-1 FUniqID;
  optional binary field_id=-1 OriginalURL;
  optional int32 field_id=-1 HID;
  optional int32 field_id=-1 IsOldCounter (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 IsEvent (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 IsParameter (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 DontCountHits (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 WithHash (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 HitColor;
  optional int64 field_id=-1 LocalEventTime;
  optional int32 field_id=-1 Age (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 Sex (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 Income (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 Interests (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 Robotness (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 RemoteIP;
  optional int32 field_id=-1 WindowName;
  optional int32 field_id=-1 OpenerName;
  optional int32 field_id=-1 HistoryLength (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 BrowserLanguage;
  optional binary field_id=-1 BrowserCountry;
  optional binary field_id=-1 SocialNetwork;
  optional binary field_id=-1 SocialAction;
  optional int32 field_id=-1 HTTPError (Int(bitWidth=16, isSigned=true));
  optional int32 field_id=-1 SendTiming;
  optional int32 field_id=-1 DNSTiming;
  optional int32 field_id=-1 ConnectTiming;
  optional int32 field_id=-1 ResponseStartTiming;
  optional int32 field_id=-1 ResponseEndTiming;
  optional int32 field_id=-1 FetchTiming;
  optional int32 field_id=-1 SocialSourceNetworkID (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 SocialSourcePage;
  optional int64 field_id=-1 ParamPrice;
  optional binary field_id=-1 ParamOrderID;
  optional binary field_id=-1 ParamCurrency;
  optional int32 field_id=-1 ParamCurrencyID (Int(bitWidth=16, isSigned=true));
  optional binary field_id=-1 OpenstatServiceName;
  optional binary field_id=-1 OpenstatCampaignID;
  optional binary field_id=-1 OpenstatAdID;
  optional binary field_id=-1 OpenstatSourceID;
  optional binary field_id=-1 UTMSource;
  optional binary field_id=-1 UTMMedium;
  optional binary field_id=-1 UTMCampaign;
  optional binary field_id=-1 UTMContent;
  optional binary field_id=-1 UTMTerm;
  optional binary field_id=-1 FromTag;
  optional int32 field_id=-1 HasGCLID (Int(bitWidth=16, isSigned=true));
  optional int64 field_id=-1 RefererHash;
  optional int64 field_id=-1 URLHash;
  optional int32 field_id=-1 CLID;
}

[raunaqmorarka]

The lack of timestamp annotation https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#timestamp in particular on eventtime, clienteventtime and localeventtime columns is a blocker for running this benchmark in Trino, Apache Hive and Apache Spark. All three engines reject attempts to read integers without timestamp logical annotation into timestamp column type.
Without the annotation, the engine cannot know what units the given data has been provided in, so it cannot scale or round values to account for any mismatch in the timestamp precision in the table definition vs in the file.

[Iskander14yo]

Currently, Spark handles this via explicit casting (like df.withColumn("EventTime", F.col("EventTime").cast("timestamp"))). While this works for Spark, its accelerators (#517) struggle since they are mostly under active development and don't support all types of casts yet. So Spark backends (e.g., Comet) will benefit from this change too.

[alexey-milovidov]

@Iskander14yo, for now, you can use one of the following workarounds:

1. Use clickhouse local to convert from the source Parquet file to a file with Date and Timestamp columns.
2. Skip failing queries and submit a pull request, so we can finish it while reviewing.

[raunaqmorarka]

@alexey-milovidov why don't you consider fixing the benchmark data by regenerating the parquet files with the expected logical annotations ?
Its a straightforward one time effort which will avoid other engines from hitting this problem and having to apply various workarounds. This issue is open for more than 3 years now.

[rschu1ze]

@raunaqmorarka ClickBench is a community project, please consider converting the Parquet files by yourself and provide them for download somewhere. We'd be happy to incorporate them into the benchmark.

[Iskander14yo]

@rschu1ze @alexey-milovidov do you want only the new Parquet file or related scripts as well? In other words, should one just attach a link or open a PR?

[rschu1ze]

It would be nice if you could place the file somewhere and adjust a couple of benchmark scripts, e.g. ClickHouse, DuckDB, Spark so it's clear how the import statements can be simplified.

[rschu1ze]

While at it, we should also make sure that the updated Parquet file has an enabled Bloom filters (writing them is supported since ClickHouse 25.2)

[alexey-milovidov]

Let me make some clarifications about the nature of this benchmark.

The benchmark is intended to test how different database management systems are able to insert and store the data and run queries.

For the purpose of testing the data insertion, we provide data in three formats: TSV, CSV, Parquet. The data in these files should have the same order, and this order should be randomized - this is important for testing, as real-life data is also often arrives out of order.

The Parquet file does not have embedded indexes, but it is also expected, because it does not matter when the data is loaded into the database in its entirety.

Some of the participants in the benchmark are not really databases, but just query engines. The first entry that we added in this category was AWS Athena, and then we added ClickHouse in its serverless (clickhouse-local) mode, and then other systems. These systems can work directly on top of the source Parquet file, but it could be suboptimal when the file does not contain indexes and is not pre-sorted.

However, if we optimize these Parquet files beforehand, it will miss the point of comparing the loading time with other databases. If it works on top of the original TSV, CSV, or Parquet files, we can confidently say that the loading time is zero. If it requires better files, like presorted ones, it hides the time that a hypothetical production system will have to spend on the sorting of the incoming data.

TLDR, How to test a system on the ideal, optimized Parquet files? - Let's say it's DataFusion. In this case, you should take the original files in the format of your choice (Parquet, CSV, TSV), transform them to the ideal files (using DataFusion as well), count it as loading time, then run the queries on top of the transformed data.