Skip to content

Commit

Permalink
Fixes and improvements. 330/346 gtests passing
Browse files Browse the repository at this point in the history
  • Loading branch information
@mhaseeb123
mhaseeb123 committed Aug 13, 2024
1 parent fb742fb commit b9067b0
Show file tree
Hide file tree
Showing 3 changed files with 67 additions and 70 deletions.
18 changes: 9 additions & 9 deletions cpp/src/io/parquet/chunk_dict.cu
View file
Original file line number Diff line number Diff line change
Expand Up @@ -41,11 +41,11 @@ struct equality_functor {
// We don't call this for nulls so this is fine
auto const equal = cudf::experimental::row::equality::nan_equal_physical_equality_comparator{};
auto const result = equal(col.element<T>(lhs_idx), col.element<T>(rhs_idx));
printf("col_type_id:%d, equality idx1:%d, idx2:%d, eq:%d\n",
/*printf("col_type_id:%d, equality idx1:%d, idx2:%d, eq:%d\n",
col.type().id(),
lhs_idx,
rhs_idx,
result);
result);*/
return result;
}
};
Expand All @@ -56,7 +56,7 @@ struct hash_functor {
__device__ auto operator()(key_type idx) const
{
auto const hashed = cudf::hashing::detail::MurmurHash3_x86_32<T>{}(col.element<T>(idx));
printf("hashing idx: %d = %d\n", idx, hashed);
// printf("hashing idx: %d = %d\n", idx, hashed);
return hashed; // cudf::hashing::detail::MurmurHash3_x86_32<T>{}(col.element<T>(idx));
}
};
Expand Down Expand Up @@ -94,7 +94,7 @@ struct map_insert_fn {
auto map_insert_ref = hash_map_ref.with_operators(cuco::insert_and_find);
// Insert
auto [iter, found] = map_insert_ref.insert_and_find(cuco::pair{i, i});
printf("Inserted k=%d, v=%d, unique=%d\n", iter->first, iter->second, found);
// printf("Inserted k=%d, v=%d, unique=%d\n", iter->first, iter->second, found);
return found;
} else {
CUDF_UNREACHABLE("Unsupported type to insert in map");
Expand Down Expand Up @@ -142,7 +142,7 @@ struct map_find_fn {
"Unable to find value in map in dictionary index construction");

// Return a pair of the found key and value.
printf("Find=%d, Found slot: k=%d, v=%d\n", i, found_slot->first, found_slot->second);
// printf("Find=%d, Found slot: k=%d, v=%d\n", i, found_slot->first, found_slot->second);
return {found_slot->first, found_slot->second};
} else {
CUDF_UNREACHABLE("Unsupported type to find in map");
Expand Down Expand Up @@ -190,8 +190,7 @@ CUDF_KERNEL void __launch_bounds__(block_size)
size_type is_unique = 0;
size_type uniq_elem_size = 0;
if (is_valid) {
auto const is_unique =
type_dispatcher(data_col.type(), map_insert_fn{storage_ref}, data_col, val_idx);
is_unique = type_dispatcher(data_col.type(), map_insert_fn{storage_ref}, data_col, val_idx);
uniq_elem_size = [&]() -> size_type {
if (not is_unique) { return 0; }
switch (col->physical_type) {
Expand Down Expand Up @@ -250,9 +249,9 @@ CUDF_KERNEL void __launch_bounds__(block_size)
[[maybe_unused]] auto const tile = cg::tiled_partition<cg_size>(cg::this_thread_block());
auto const t = cg::this_thread_block().thread_rank();

__shared__ cuda::atomic<size_type, cuda::thread_scope_block> counter;
__shared__ cuda::atomic<size_type, SCOPE> counter;
using cuda::std::memory_order_relaxed;
if (t == 0) { new (&counter) cuda::atomic<size_type, cuda::thread_scope_block>{0}; }
if (t == 0) { new (&counter) cuda::atomic<size_type, SCOPE>{0}; }
__syncthreads();

for (size_type i = 0; i < chunk.dict_map_size; i += block_size) {
Expand All @@ -267,6 +266,7 @@ CUDF_KERNEL void __launch_bounds__(block_size)
// If sorting dict page ever becomes a hard requirement, enable the following statement and
// add a dict sorting step before storing into the slot's second field.
// chunk.dict_data_idx[loc] = t + i;
// printf("Replacing slot->data()->second: %d, %d\n", slot->data()->second, loc);
slot->data()->second = loc;
}
}
Expand Down
20 changes: 8 additions & 12 deletions cpp/src/io/parquet/writer_impl.cu
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1322,18 +1322,13 @@ build_chunk_dictionaries(hostdevice_2dvector<EncColumnChunk>& chunks,
map_storage.initialize(cuco::pair{KEY_SENTINEL, VALUE_SENTINEL},
cuda::stream_ref{stream.value()});

std::cout << "Offsets: " << h_chunks.size() << std::endl;

std::for_each(thrust::make_counting_iterator(static_cast<std::size_t>(0)),
thrust::make_counting_iterator(h_chunks.size()),
[&](auto const idx) {
auto& chunk = h_chunks[idx];
if (chunk.use_dictionary) {
chunk.dict_map_offset = map_offsets[idx];
std::cout << "off: " << map_offsets[idx] << ", size: " << chunk.dict_map_size
<< std::endl;
}
});
std::for_each(
thrust::make_zip_iterator(thrust::make_tuple(h_chunks.begin(), map_offsets.begin())),
thrust::make_zip_iterator(thrust::make_tuple(h_chunks.end(), map_offsets.end())),
[&](auto elem) -> void {
auto& chunk = thrust::get<0>(elem);
if (chunk.use_dictionary) { chunk.dict_map_offset = thrust::get<1>(elem); }
});

chunks.host_to_device_async(stream);

Expand Down Expand Up @@ -1400,6 +1395,7 @@ build_chunk_dictionaries(hostdevice_2dvector<EncColumnChunk>& chunks,
chunks.host_to_device_async(stream);
collect_map_entries(map_storage.data(), chunks.device_view().flat_view(), stream);
get_dictionary_indices(map_storage.data(), frags, stream);
chunks.device_to_host_async(stream);

return std::pair(std::move(dict_data), std::move(dict_index));
}
Expand Down
99 changes: 50 additions & 49 deletions cpp/tests/io/parquet_writer_test.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ void test_durations(mask_op_t mask_op, bool use_byte_stream_split, bool arrow_sc
{
std::default_random_engine generator;
std::uniform_int_distribution<int> distribution_d(0, 30);
[[maybe_unused]]auto sequence_d = cudf::detail::make_counting_transform_iterator(
auto sequence_d = cudf::detail::make_counting_transform_iterator(
0, [&](auto i) { return distribution_d(generator); });

std::uniform_int_distribution<int> distribution_s(0, 86400);
Expand All @@ -52,8 +52,8 @@ void test_durations(mask_op_t mask_op, bool use_byte_stream_split, bool arrow_sc

auto mask = cudf::detail::make_counting_transform_iterator(0, mask_op);

constexpr auto num_rows = 5650; // WORKS UNTIL 5649, fails beyond 5650
// Durations longer than a day are not exactly valid, but cudf should be able to round trip
constexpr auto num_rows = 100;
// Durations longer than a day are not exactly valid, but cudf should be able to round trip
auto durations_d = cudf::test::fixed_width_column_wrapper<cudf::duration_D, int64_t>(
sequence_d, sequence_d + num_rows, mask);
auto durations_s = cudf::test::fixed_width_column_wrapper<cudf::duration_s, int64_t>(
Expand All @@ -65,7 +65,7 @@ void test_durations(mask_op_t mask_op, bool use_byte_stream_split, bool arrow_sc
auto durations_ns = cudf::test::fixed_width_column_wrapper<cudf::duration_ns, int64_t>(
sequence, sequence + num_rows, mask);

auto expected = table_view{{/*durations_d, */durations_s,/*durations_ms, durations_us, durations_ns*/}};
auto expected = table_view{{durations_d, durations_s, durations_ms, durations_us, durations_ns}};

if (use_byte_stream_split) {
cudf::io::table_input_metadata expected_metadata(expected);
Expand All @@ -85,35 +85,35 @@ void test_durations(mask_op_t mask_op, bool use_byte_stream_split, bool arrow_sc
cudf::io::parquet_reader_options::builder(cudf::io::source_info{filepath})
.use_arrow_schema(arrow_schema);
auto result = cudf::io::read_parquet(in_opts);
/*

auto durations_d_got =
cudf::cast(result.tbl->view().column(0), cudf::data_type{cudf::type_id::DURATION_DAYS});
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_d, durations_d_got->view());*/
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_d, durations_d_got->view());

if (arrow_schema) {
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_s, result.tbl->view().column(1));
} else {
auto durations_s_got =
cudf::cast(result.tbl->view().column(1), cudf::data_type{cudf::type_id::DURATION_SECONDS});
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_s, durations_s_got->view());
}

if (arrow_schema) {
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_s, result.tbl->view().column(0));
} else {
auto durations_s_got =
cudf::cast(result.tbl->view().column(0), cudf::data_type{cudf::type_id::DURATION_SECONDS});
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_s, durations_s_got->view());
}
/*
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_ms, result.tbl->view().column(1));
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_us, result.tbl->view().column(2));
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_ns, result.tbl->view().column(3));*/
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_ms, result.tbl->view().column(2));
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_us, result.tbl->view().column(3));
CUDF_TEST_EXPECT_COLUMNS_EQUAL(durations_ns, result.tbl->view().column(4));
}

TEST_F(ParquetWriterTest, Durations)
{
test_durations([](auto i) { return true; }, false, false);
// test_durations([](auto i) { return (i % 2) != 0; }, false, false);
// test_durations([](auto i) { return (i % 3) != 0; }, false, false);
// test_durations([](auto i) { return false; }, false, false);
test_durations([](auto i) { return (i % 2) != 0; }, false, false);
test_durations([](auto i) { return (i % 3) != 0; }, false, false);
test_durations([](auto i) { return false; }, false, false);

// test_durations([](auto i) { return true; }, false, true);
// test_durations([](auto i) { return (i % 2) != 0; }, false, true);
// test_durations([](auto i) { return (i % 3) != 0; }, false, true);
// test_durations([](auto i) { return false; }, false, true);
test_durations([](auto i) { return true; }, false, true);
test_durations([](auto i) { return (i % 2) != 0; }, false, true);
test_durations([](auto i) { return (i % 3) != 0; }, false, true);
test_durations([](auto i) { return false; }, false, true);
}

TEST_F(ParquetWriterTest, MultiIndex)
Expand Down Expand Up @@ -2007,7 +2007,7 @@ TEST_F(ParquetWriterTest, WriteFixedLenByteArray)
srand(31337);
using cudf::io::parquet::detail::Encoding;
constexpr int fixed_width = 16;
constexpr cudf::size_type num_rows = 200;
constexpr cudf::size_type num_rows = 10;
std::vector<uint8_t> data(num_rows * fixed_width);
std::vector<cudf::size_type> offsets(num_rows + 1);

Expand All @@ -2025,25 +2025,26 @@ TEST_F(ParquetWriterTest, WriteFixedLenByteArray)
auto off_child = cudf::test::fixed_width_column_wrapper<int32_t>(offsets.begin(), offsets.end());
auto col = cudf::make_lists_column(num_rows, off_child.release(), data_child.release(), 0, {});

auto expected = table_view{{*col, *col, *col, *col}};
auto expected = table_view{{/**col, *col, *col,*/ *col}};
cudf::io::table_input_metadata expected_metadata(expected);

/*
expected_metadata.column_metadata[0]
.set_name("flba_plain")
.set_type_length(fixed_width)
.set_encoding(cudf::io::column_encoding::PLAIN)
.set_output_as_binary(true);
expected_metadata.column_metadata[1]
.set_name("flba_split")
.set_type_length(fixed_width)
.set_encoding(cudf::io::column_encoding::BYTE_STREAM_SPLIT)
.set_output_as_binary(true);
expected_metadata.column_metadata[2]
.set_name("flba_delta")
.set_type_length(fixed_width)
.set_encoding(cudf::io::column_encoding::DELTA_BYTE_ARRAY)
.set_output_as_binary(true);*/
expected_metadata.column_metadata[0]
.set_name("flba_plain")
.set_type_length(fixed_width)
.set_encoding(cudf::io::column_encoding::PLAIN)
.set_output_as_binary(true);
expected_metadata.column_metadata[1]
.set_name("flba_split")
.set_type_length(fixed_width)
.set_encoding(cudf::io::column_encoding::BYTE_STREAM_SPLIT)
.set_output_as_binary(true);
expected_metadata.column_metadata[2]
.set_name("flba_delta")
.set_type_length(fixed_width)
.set_encoding(cudf::io::column_encoding::DELTA_BYTE_ARRAY)
.set_output_as_binary(true);
expected_metadata.column_metadata[3]
.set_name("flba_dict")
.set_type_length(fixed_width)
.set_encoding(cudf::io::column_encoding::DICTIONARY)
Expand All @@ -2067,7 +2068,7 @@ TEST_F(ParquetWriterTest, WriteFixedLenByteArray)
read_footer(source, &fmd);

// check that the schema retains the FIXED_LEN_BYTE_ARRAY type
for (int i = 1; i <= 4; i++) {
for (int i = 1; i <= 1; i++) {
EXPECT_EQ(fmd.schema[i].type, cudf::io::parquet::detail::Type::FIXED_LEN_BYTE_ARRAY);
EXPECT_EQ(fmd.schema[i].type_length, fixed_width);
}
Expand All @@ -2078,14 +2079,14 @@ TEST_F(ParquetWriterTest, WriteFixedLenByteArray)
};

// requested plain
/* expect_enc(0, Encoding::PLAIN);
// requested byte_stream_split
expect_enc(1, Encoding::BYTE_STREAM_SPLIT);
// requested delta_byte_array
expect_enc(2, Encoding::DELTA_BYTE_ARRAY);
// requested dictionary, but should fall back to plain
// TODO: update if we get FLBA working with dictionary encoding*/
expect_enc(0, Encoding::PLAIN);
// requested byte_stream_split
expect_enc(1, Encoding::BYTE_STREAM_SPLIT);
// requested delta_byte_array
expect_enc(2, Encoding::DELTA_BYTE_ARRAY);
// requested dictionary, but should fall back to plain
// TODO: update if we get FLBA working with dictionary encoding
expect_enc(3, Encoding::PLAIN);
}

/////////////////////////////////////////////////////////////
Expand Down

0 comments on commit b9067b0

Please sign in to comment.