diff --git a/velox/connectors/clp/ClpDataSource.cpp b/velox/connectors/clp/ClpDataSource.cpp index 6bc4d1d1414..d41c69b99c1 100644 --- a/velox/connectors/clp/ClpDataSource.cpp +++ b/velox/connectors/clp/ClpDataSource.cpp @@ -93,6 +93,9 @@ void ClpDataSource::addFieldsRecursively( case TypeKind::ARRAY: clpColumnType = search_lib::ColumnType::Array; break; + case TypeKind::TIMESTAMP: + clpColumnType = search_lib::ColumnType::Timestamp; + break; default: VELOX_USER_FAIL("Type not supported: {}", columnType->name()); } diff --git a/velox/connectors/clp/search_lib/ClpCursor.cpp b/velox/connectors/clp/search_lib/ClpCursor.cpp index bab940c92c3..fa6d67bf22b 100644 --- a/velox/connectors/clp/search_lib/ClpCursor.cpp +++ b/velox/connectors/clp/search_lib/ClpCursor.cpp @@ -213,6 +213,11 @@ ErrorCode ClpCursor::loadArchive() { case ColumnType::Array: columnDescriptor->set_matching_types(LiteralType::ArrayT); break; + case ColumnType::Timestamp: + columnDescriptor->set_matching_types( + LiteralType::EpochDateT | LiteralType::IntegerT | + LiteralType::FloatT); + break; default: break; } diff --git a/velox/connectors/clp/search_lib/ClpCursor.h b/velox/connectors/clp/search_lib/ClpCursor.h index 43db49c7841..f6303a2ef94 100644 --- a/velox/connectors/clp/search_lib/ClpCursor.h +++ b/velox/connectors/clp/search_lib/ClpCursor.h @@ -49,7 +49,15 @@ enum class ErrorCode { Success }; -enum class ColumnType { String, Integer, Float, Array, Boolean, Unknown = -1 }; +enum class ColumnType { + Array, + Boolean, + Float, + Integer, + String, + Timestamp, + Unknown = -1 +}; struct Field { ColumnType type; diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp index be73e842d73..6bd58e8a2a2 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.cpp +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.cpp @@ -14,16 +14,113 @@ * limitations under the License. */ +#include +#include #include #include "clp_s/ColumnReader.hpp" +#include "clp_s/SchemaTree.hpp" #include "velox/connectors/clp/search_lib/ClpVectorLoader.h" +#include "velox/type/Timestamp.h" #include "velox/vector/ComplexVector.h" #include "velox/vector/FlatVector.h" namespace facebook::velox::connector::clp::search_lib { +namespace { + +enum class TimestampPrecision : uint8_t { + Seconds, + Milliseconds, + Microseconds, + Nanoseconds +}; + +/// Estimates the precision of an epoch timestamp as seconds, milliseconds, +/// microseconds, or nanoseconds. +/// +/// This heuristic relies on the fact that 1 year of epoch nanoseconds is +/// approximately 1000 years of epoch microseconds and so on. This heuristic +/// can be unreliable for timestamps sufficiently close to the epoch, but +/// should otherwise be accurate for the next 1000 years. +/// +/// Note: Future versions of the clp-s archive format will adopt a +/// nanosecond-precision integer timestamp format (as opposed to the current +/// format which allows other precisions), at which point we can remove this +/// heuristic. +/// +/// @param timestamp +/// @return the estimated timestamp precision +template +auto estimatePrecision(T timestamp) -> TimestampPrecision { + constexpr int64_t kEpochMilliseconds1971{31536000000}; + constexpr int64_t kEpochMicroseconds1971{31536000000000}; + constexpr int64_t kEpochNanoseconds1971{31536000000000000}; + auto absTimestamp = timestamp >= 0 ? timestamp : -timestamp; + + if (absTimestamp > kEpochNanoseconds1971) { + return TimestampPrecision::Nanoseconds; + } else if (absTimestamp > kEpochMicroseconds1971) { + return TimestampPrecision::Microseconds; + } else if (absTimestamp > kEpochMilliseconds1971) { + return TimestampPrecision::Milliseconds; + } else { + return TimestampPrecision::Seconds; + } +} + +auto convertToVeloxTimestamp(double timestamp) -> Timestamp { + switch (estimatePrecision(timestamp)) { + case TimestampPrecision::Nanoseconds: + timestamp /= Timestamp::kNanosInSecond; + break; + case TimestampPrecision::Microseconds: + timestamp /= Timestamp::kMicrosecondsInSecond; + break; + case TimestampPrecision::Milliseconds: + timestamp /= Timestamp::kMillisecondsInSecond; + break; + case TimestampPrecision::Seconds: + break; + } + double seconds{std::floor(timestamp)}; + double nanoseconds{(timestamp - seconds) * Timestamp::kNanosInSecond}; + return Timestamp( + static_cast(seconds), static_cast(nanoseconds)); +} + +auto convertToVeloxTimestamp(int64_t timestamp) -> Timestamp { + int64_t precisionDifference{Timestamp::kNanosInSecond}; + switch (estimatePrecision(timestamp)) { + case TimestampPrecision::Nanoseconds: + break; + case TimestampPrecision::Microseconds: + precisionDifference = + Timestamp::kNanosInSecond / Timestamp::kNanosecondsInMicrosecond; + break; + case TimestampPrecision::Milliseconds: + precisionDifference = + Timestamp::kNanosInSecond / Timestamp::kNanosecondsInMillisecond; + break; + case TimestampPrecision::Seconds: + precisionDifference = + Timestamp::kNanosInSecond / Timestamp::kNanosInSecond; + break; + } + int64_t seconds{timestamp / precisionDifference}; + int64_t nanoseconds{ + (timestamp % precisionDifference) * + (Timestamp::kNanosInSecond / precisionDifference)}; + if (nanoseconds < 0) { + seconds -= 1; + nanoseconds += Timestamp::kNanosInSecond; + } + return Timestamp(seconds, static_cast(nanoseconds)); +} + +} // namespace + ClpVectorLoader::ClpVectorLoader( clp_s::BaseColumnReader* columnReader, ColumnType nodeType, @@ -57,6 +154,51 @@ void ClpVectorLoader::populateData(RowSet rows, VectorPtr vector) { } } +template +void ClpVectorLoader::populateTimestampData( + RowSet rows, + FlatVector* vector) { + bool supportedNodeType{false}; + switch (Type) { + case clp_s::NodeType::Float: + case clp_s::NodeType::Integer: + case clp_s::NodeType::DateString: + supportedNodeType = true; + break; + default: + break; + } + if (columnReader_ == nullptr || false == supportedNodeType) { + for (int vectorIndex : rows) { + vector->setNull(vectorIndex, true); + } + return; + } + + for (int vectorIndex : rows) { + auto messageIndex = (*filteredRowIndices_)[vectorIndex]; + + if (clp_s::NodeType::Float == Type) { + auto reader = static_cast(columnReader_); + vector->set( + vectorIndex, + convertToVeloxTimestamp( + std::get(reader->extract_value(messageIndex)))); + } else if (clp_s::NodeType::Integer == Type) { + auto reader = static_cast(columnReader_); + vector->set( + vectorIndex, + convertToVeloxTimestamp( + std::get(reader->extract_value(messageIndex)))); + } else { + auto reader = static_cast(columnReader_); + vector->set( + vectorIndex, + convertToVeloxTimestamp(reader->get_encoded_time(messageIndex))); + } + } +} + void ClpVectorLoader::loadInternal( RowSet rows, ValueHook* hook, @@ -142,6 +284,23 @@ void ClpVectorLoader::loadInternal( } break; } + case ColumnType::Timestamp: { + auto timestampVector = vector->asFlatVector(); + if (nullptr != dynamic_cast(columnReader_)) { + populateTimestampData(rows, timestampVector); + } else if ( + nullptr != + dynamic_cast(columnReader_)) { + populateTimestampData( + rows, timestampVector); + } else if ( + nullptr != dynamic_cast(columnReader_)) { + populateTimestampData(rows, timestampVector); + } else { + populateTimestampData(rows, timestampVector); + } + break; + } default: VELOX_FAIL("Unsupported column type"); } @@ -160,5 +319,18 @@ template void ClpVectorLoader::populateData( template void ClpVectorLoader::populateData( RowSet rows, FlatVector* vector); +template void ClpVectorLoader::populateTimestampData( + RowSet rows, + FlatVector* vector); +template void ClpVectorLoader::populateTimestampData( + RowSet rows, + FlatVector* vector); +template void +ClpVectorLoader::populateTimestampData( + RowSet rows, + FlatVector* vector); +template void ClpVectorLoader::populateTimestampData( + RowSet rows, + FlatVector* vector); } // namespace facebook::velox::connector::clp::search_lib diff --git a/velox/connectors/clp/search_lib/ClpVectorLoader.h b/velox/connectors/clp/search_lib/ClpVectorLoader.h index ab57b52dfd9..36af6d7b807 100644 --- a/velox/connectors/clp/search_lib/ClpVectorLoader.h +++ b/velox/connectors/clp/search_lib/ClpVectorLoader.h @@ -16,7 +16,12 @@ #pragma once +#include "clp_s/ColumnReader.hpp" +#include "clp_s/SchemaTree.hpp" + #include "velox/connectors/clp/search_lib/ClpCursor.h" +#include "velox/type/Timestamp.h" +#include "velox/vector/FlatVector.h" #include "velox/vector/LazyVector.h" namespace clp_s { @@ -45,6 +50,11 @@ class ClpVectorLoader : public VectorLoader { template void populateData(RowSet rows, VectorPtr vector); + template + void populateTimestampData( + RowSet rows, + FlatVector* vector); + clp_s::BaseColumnReader* columnReader_; ColumnType nodeType_; std::shared_ptr> filteredRowIndices_; diff --git a/velox/connectors/clp/tests/ClpConnectorTest.cpp b/velox/connectors/clp/tests/ClpConnectorTest.cpp index bb351984b64..cea13f119f2 100644 --- a/velox/connectors/clp/tests/ClpConnectorTest.cpp +++ b/velox/connectors/clp/tests/ClpConnectorTest.cpp @@ -26,6 +26,8 @@ #include "velox/exec/tests/utils/AssertQueryBuilder.h" #include "velox/exec/tests/utils/OperatorTestBase.h" #include "velox/exec/tests/utils/PlanBuilder.h" +#include "velox/type/Timestamp.h" +#include "velox/type/Type.h" namespace { @@ -34,6 +36,10 @@ using namespace facebook::velox::connector::clp; using facebook::velox::exec::test::PlanBuilder; +// Epoch seconds and nanoseconds for the timestamp "2025-04-30T08:50:05Z" +constexpr int64_t kTestTimestampSeconds{1746003005}; +constexpr uint64_t kTestTimestampNanoseconds{0ULL}; + class ClpConnectorTest : public exec::test::OperatorTestBase { public: const std::string kClpConnectorId = "test-clp"; @@ -165,7 +171,7 @@ TEST_F(ClpConnectorTest, test2NoPushdown) { .startTableScan() .outputType( ROW({"timestamp", "event"}, - {VARCHAR(), + {TIMESTAMP(), ROW({"type", "subtype", "severity"}, {VARCHAR(), VARCHAR(), VARCHAR()})})) .tableHandle(std::make_shared( @@ -173,7 +179,7 @@ TEST_F(ClpConnectorTest, test2NoPushdown) { .assignments( {{"timestamp", std::make_shared( - "timestamp", "timestamp", VARCHAR(), true)}, + "timestamp", "timestamp", TIMESTAMP(), true)}, {"event", std::make_shared( "event", @@ -192,7 +198,8 @@ TEST_F(ClpConnectorTest, test2NoPushdown) { getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))}); auto expected = makeRowVector({// timestamp - makeFlatVector({"2025-04-30T08:50:05Z"}), + makeFlatVector({Timestamp( + kTestTimestampSeconds, kTestTimestampNanoseconds)}), // event makeRowVector({ // event.type @@ -211,7 +218,7 @@ TEST_F(ClpConnectorTest, test2Pushdown) { .startTableScan() .outputType( ROW({"timestamp", "event"}, - {VARCHAR(), + {TIMESTAMP(), ROW({"type", "subtype", "severity"}, {VARCHAR(), VARCHAR(), VARCHAR()})})) .tableHandle(std::make_shared( @@ -224,7 +231,7 @@ TEST_F(ClpConnectorTest, test2Pushdown) { .assignments( {{"timestamp", std::make_shared( - "timestamp", "timestamp", VARCHAR(), true)}, + "timestamp", "timestamp", TIMESTAMP(), true)}, {"event", std::make_shared( "event", @@ -239,7 +246,8 @@ TEST_F(ClpConnectorTest, test2Pushdown) { getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))}); auto expected = makeRowVector({// timestamp - makeFlatVector({"2025-04-30T08:50:05Z"}), + makeFlatVector({Timestamp( + kTestTimestampSeconds, kTestTimestampNanoseconds)}), // event makeRowVector({ // event.type @@ -258,7 +266,7 @@ TEST_F(ClpConnectorTest, test2Hybrid) { .startTableScan() .outputType( ROW({"timestamp", "event"}, - {VARCHAR(), + {TIMESTAMP(), ROW({"type", "subtype", "severity", "tags"}, {VARCHAR(), VARCHAR(), VARCHAR(), ARRAY(VARCHAR())})})) .tableHandle(std::make_shared( @@ -270,7 +278,7 @@ TEST_F(ClpConnectorTest, test2Hybrid) { .assignments( {{"timestamp", std::make_shared( - "timestamp", "timestamp", VARCHAR(), true)}, + "timestamp", "timestamp", TIMESTAMP(), true)}, {"event", std::make_shared( "event", @@ -286,7 +294,8 @@ TEST_F(ClpConnectorTest, test2Hybrid) { getResults(plan, {makeClpSplit(getExampleFilePath("test_2.clps"))}); auto expected = makeRowVector( {// timestamp - makeFlatVector({"2025-04-30T08:50:05Z"}), + makeFlatVector( + {Timestamp(kTestTimestampSeconds, kTestTimestampNanoseconds)}), // event makeRowVector({// event.type makeFlatVector({"storage"}), @@ -302,6 +311,32 @@ TEST_F(ClpConnectorTest, test2Hybrid) { test::assertEqualVectors(expected, output); } +TEST_F(ClpConnectorTest, test3TimestampMarshalling) { + auto plan = PlanBuilder(pool_.get()) + .startTableScan() + .outputType(ROW({"timestamp"}, {TIMESTAMP()})) + .tableHandle(std::make_shared( + kClpConnectorId, "test_3", nullptr)) + .assignments( + {{"timestamp", + std::make_shared( + "timestamp", "timestamp", TIMESTAMP(), true)}}) + .endTableScan() + .planNode(); + + auto output = + getResults(plan, {makeClpSplit(getExampleFilePath("test_3.clps"))}); + auto expected = makeRowVector({ + // timestamp + makeFlatVector( + {Timestamp(kTestTimestampSeconds, kTestTimestampNanoseconds), + Timestamp(kTestTimestampSeconds, kTestTimestampNanoseconds), + Timestamp(kTestTimestampSeconds, kTestTimestampNanoseconds), + Timestamp(kTestTimestampSeconds, kTestTimestampNanoseconds)}), + }); + test::assertEqualVectors(expected, output); +} + } // namespace int main(int argc, char** argv) { diff --git a/velox/connectors/clp/tests/examples/test_2.clps b/velox/connectors/clp/tests/examples/test_2.clps index ad773162d3a..22ddd92bb5b 100644 Binary files a/velox/connectors/clp/tests/examples/test_2.clps and b/velox/connectors/clp/tests/examples/test_2.clps differ diff --git a/velox/connectors/clp/tests/examples/test_3.clps b/velox/connectors/clp/tests/examples/test_3.clps new file mode 100644 index 00000000000..3dd9f37b4ef Binary files /dev/null and b/velox/connectors/clp/tests/examples/test_3.clps differ diff --git a/velox/connectors/clp/tests/examples/test_3.ndjson b/velox/connectors/clp/tests/examples/test_3.ndjson new file mode 100644 index 00000000000..0839ab1263b --- /dev/null +++ b/velox/connectors/clp/tests/examples/test_3.ndjson @@ -0,0 +1,4 @@ +{"timestamp": "2025-04-30T08:50:05.000Z"} +{"timestamp": 1746003005.0} +{"timestamp": 1746003005000000} +{"timestamp": 1746003005000000000}