Collect column statistics on write [v2]

[arhimondr]

Important changes since the version 1:

• Commits from Return HiveColumnStatistics from the HiveMetastore interface to Move createPartitionValues method to a utility class were extracted into a separate PR and merged: Preliminary column statistics refactorings #10972
• Added Implement AutoCloseableCloser commit.
• SPI bits extracted to Collect column statistics on table write: SPI
• getInsertStatisticsMetadata and getNewTableStatisticsMetadata method were merged into the single getStatisticsCollectionMetadata method
• AggregationOperator integration with the TableWriterOperator has changed
• ExtendedHiveMetastore#isColumnStatistitcsSupported() replaced with the ExtendedHiveMetastore#getSupportedColumnStatistics. That eliminates a need of CollectibleStatisticsProvider. Commit: Replace supportsColumnStatistics with getSupportedColumnStatistics
• ENABLED_FOR_MARKED_TABLES option has been removed as well as a related table property.
• Migrate column statistics on drop and rename column commit has been dropped
• Added Collect column statistics on table write: Smoke Tests commit

[arhimondr]

@electrum I'm still working on the integration test

[arhimondr]

Please ignore the Fix memory leaks in tests from the presto-tests module commit, it is part of the #11062 PR

[findepi]

public <T extends AutoCloseable> T register(T closeable)

[findepi]

Guava Closer (which this class mimics) closes in reverse order.
Also, it doesn't close the underlying resource twice, even if close is called twice (eg concurrently)

[arhimondr]

I checked that. First i didn't get it, but now i think i got it.

Given the example

stream = closer.register(new OutputStream)
writer = closer.register(new OutputWriter(stream))

it makes total sense

[arhimondr]

even if close is called twice (eg concurrently)

The Closer is not thread safe. It is hard to reason what is going on when using it concurrently.

[findepi]

if we reach here, throw new Error or AssertionError would be more appropriate.. since this is unreachable

[findepi]

replace those 2 with propagateIfPossible(rootCause, Exception.class)

[arhimondr]

We use throwIfInstanceOf across the code

[electrum]

Almost all usages of throwIfInstanceOf are alone. The few places that it is combined with throwIfUnchecked should be converted to propagateIfPossible since that is more succinct.

[findepi]

only new tables? I think also updated ones

[findepi]

again -- "new"?

[findepi]

typo: groping
(field & ctor & builder too)

[arhimondr]

Ditto

[findepi]

Did you mean new ArrayList<>(groupingColumns)?
Otherwise this is redundant (ctor does that) and allows creating mutable ComputedStatistics objects

[arhimondr]

Removed the unmodifiable* at all here. Since those are in the constructor.

[findepi]

convert ctor to builder methods withGroupingColumns (be sure to make defensive copy there)

[arhimondr]

This field are required. Why do you want to make them builder methods? Also defensive copy is made in the constructor. Why make it twice?

[findepi]

unmodifiableList( new ArrayList<>( requireNonNull( ....

[arhimondr]

I messed up during the rebase. That was applied in the commit later. Moving it here.

[findepi]

"Collect column statistics on table write: Planner" -- skimmed only. I refuse to understand SymbolMapper for now

[findepi]

nit: in case this fails some day, including full collections in the message might be helpful (perhaps even instead of the sizes):

verify(columnNames.size() == symbols.size(), "columnNames.size() != symbols.size(): %s and %s", columnNames, symbols);

[findepi]

You should merge this if into the for loop above:

for (TableStatisticType type : statisticsMetadata.getTableStatistics()) {
    if (type != ROW_COUNT) {
       ...
    }

    // plan the row count agg
}

[kokosing]

Or:

for (TableStatisticType type : statisticsMetadata.getTableStatistics()) {
    if (type == ROW_COUNT) {
       // plan the row count agg.
    }
   else {
       // fail
   }
}

[findepi]

line above uses verify, this one uses requireNonNull for checking nullity, but i see no reason for it

[arhimondr]

Replaced with a verify. Just in case the type for the symbol is not there.

[findepi]

what would happen if you removed this check (and the type wasn't orderable)?
would we get some useful enough exception during createAggregation in next line?

[arhimondr]

I though that the explicit message might be more clear (in case someone ever wanted to compute MIN/MAX statistics for the type that is not orderable).

[findepi]

nit: these {, } are redundant in most of the cases and affect readability. IMHO it's better without them, but no strong opinion here

[arhimondr]

I usually go without { } if the switch case blocks are one-liners. I like them for multiliners though.

[findepi]

"Collect column statistics on table write: Execution"

[findepi]

unless it is full

or there is some unfinishedWork

[findepi]

typo: ans

[findepi]

com.facebook.presto.spi.Page#getPositions ?
else, leave a comment why not using this

[arhimondr]

Copy just seems to be safer. Because it is copy. There will be not that much data to copy. And in most of the cases it will return the page unaltered. But it can be get as well. I don't have a strong opinion here.

[findepi]

Well -- we sometimes return the whole page (few lines earlier), so there cannot be assumption that you do a copy.

(no strong opinion either, just opportunity to simplify the code)

[findepi]

nit: move after if (state != State.FINISHING) { .. }

[arhimondr]

In most of the implementation isDone is a simple flag check. But i don't mind changing.

[findepi]

verify(statisticsAggregation.isBlocked().isDone()); to fail fast rather than voidly looping in case something goes wrong

[findepi]

They may go out of sync. Why not make a defensive copy?

[findepi]

remove if, else's code is generic enough.

if you don't want to go through com.google.common.util.concurrent.Futures#allAsList(com.google.common.util.concurrent.ListenableFuture<? extends V>...) in all-done case, create a helper method that does the if

[findepi]

check state first (cheaper first)

[findepi]

you should yield here (return). Operator shouldn't do a ton of work within single call, otherwise a query might be "unkillable"

[arhimondr]

IRL we cannot get here more data than a single page. We group the statistics on per-partition basis. And we never insert more than 100 partitions at once by default. But even if we inserted 100_000 - there still will be not enough data for more than a several pages. But than you would definitely have more severe problems. So for the sake of code simplicity in this class (which is already complex), i would go with what we have now.

[findepi]

fine, but please add a comment that we deliberately not yield here

[findepi]

wrap in try-with-r

[arhimondr]

It is unnecessary. We don't expect the register() methods to fail.

[findepi]

yea... it's also customary. just a matter a taste

[rschlussel]

should these match what's in the ColumnStatistics spi? I think it would make sense for the set of stats we support reading and writing to be the same. If so, there's no max/average_value_size_in_bytes, but there is a total_size_in_bytes.

[findepi]

not necessarily. SPI is a common denominator, something we ingest. Here we have superset of all possible stats external systems can store.
SPI says want we want for CBO.
Here we say what hive need (or other programs using metastore may need, including ourselves)

[arhimondr]

@rschlussel There is no exact match. NUMBER_OF_TRUE_VALUES is something boolean specific, NUMBER_OF_NON_NULL_VALUES is chosen over NUMBER_OF_NULL_VALUES just because it is easier to compute (no extra projection), and so on.

[rschlussel]

s/ans/and

[rschlussel]

can you add a comment somewhere explaining the multiplexing that you explained to me in person?

[rschlussel]

These are the stats channels, right? Could you call them STATISTICS_CHANNELS

[arhimondr]

This is a total count of WRITER_CHANNELS.

[rschlussel]

you didn't actually touch this file, right? just reformatted/added this null check. You could extract and merge this separately.

[arhimondr]

No idea why is this change here, and why would i even tough this class here. I extracted it to a separate commit.

[rschlussel]

does this need to be both or neither? Sort of makes sense that they'd both be required fields, but it's not totally useless to have only one. E.g. if min for x was 10 and you had a predicate where x < 8, you'd know that nothing matches.

[arhimondr]

We always compute MIN/MAX in pairs. Checking this in a single if for simplicity. Will add a assertion though.

[rschlussel]

do we normally have tests that our session properties are actually gating the things we think they do? The test is fine, but curious why for this?

[arhimondr]

Before that there was more complicated logic involving both, table and session property. Since now the logic is trivial i will just remove this test.

[kokosing]

any tests?

[arhimondr]

This was needed for the approx_distinct.

Actually the equals semantics for CHAR in Presto are not correct. Because of the binary comparison, the EQUALS operator would return false for abc and abc. That will result into storing wrong statistics for the CHAR type. For now i'm not going to collect the NDV statistic for the CHAR type, untils the CHAR semantics are fixed.

[arhimondr]

UPD: It is not possible to save CHAR statistics without setting NDV. Going to go with the return XxHash64.hash(slice); implementation. But potentially it may return higher number of distinct values.

[arhimondr]

#11101

[findepi]

Actually the equals semantics for CHAR in Presto are not correct. Because of the binary comparison, the EQUALS operator would return false for abc⎵⎵ and abc.

as noted #11101 (comment), internal repr is normalized, with trailing spaces removed (wheever there is a trailing space, this is a bug)

[kokosing]

should not be this private if you provide a builder

[kokosing]

private?

[kokosing]

please chain this:

List<..> outputs = builder().add().add().build()

[kokosing]

inline writerOutputs

[arhimondr]

It is used in multiple places (:405)

[kokosing]

Or:

for (TableStatisticType type : statisticsMetadata.getTableStatistics()) {
    if (type == ROW_COUNT) {
       // plan the row count agg.
    }
   else {
       // fail
   }
}

[kokosing]

canonicalizePartitioningScheme? Or maybe even just canonicalize?

[kokosing]

should not you use mapAndDistinct here?

[kokosing]

statisticsAggregationOperator?

[arhimondr]

Didn't want to make it oververbose, but it looks like it indeed decreases readablity.

[arhimondr]

Renamed statisticsAggregationOperatorFactory as well

[kokosing]

can you throw SkipException here and elsewhere?

[findepi]

it isn't our convention to do so

[kokosing]

It is not, but I think it should be. Telling people that test is passing while it is not possible for test to pass might be a bit misleading. I hope raising SkipException might have a better developer experience.

[findepi]

This is provided that someone reads the list of passing tests. I don't.
I do read lists of failing tests only. And Jenkins's testng plugin lists all the skipped tests, which isn't entirely interesting when the test is skipped because the functionality simply does not exist

[arhimondr]

Usually we throw SkipException if there is no easy way of disabling a test in any other way. (throwing it from somewhere deep inside the mock objects)

[findepi]

"Collect column statistics on table write: Hive Connector"

[findepi]

Hive stores MIN,MAX as well -- we don't use this currently, but we may in the future. Also, other tools may make use of this.
Consider // TODO ...

[findepi]

add // TODO #7122 support non-legacy TIMESTAMP

[findepi]

There are types that are not here because they are not supported by Hive connector. eg. TIME_WITH_TIME_ZONE, TIMESTAMP_WITH_TIME_ZONE.
I think it would be better to end this method with

// Throwing here to make sure this method is updated when a new type is added in Hive connector
throw new IllegalArgumentException("Unsupported type: " + type);

[arhimondr]

There are many types that are not here. For example ARRAY/MAP/ROW. If i throw exception here i would need to check if the type is supported above.

[findepi]

i still would prefer a whitelist approach. Are there types to add other than array/map/row?

[arhimondr]

No. Yeah, maybe you are right, maybe we should return ImmutableList.empty() for that types, and throw an exception otherwise.

[findepi]

MIN,MAX as well?

Isn't AVERAGE_VALUE_SIZE_IN_BYTES interesting? (with trailing spaces trimmed). Add a TODO

[arhimondr]

Isn't AVERAGE_VALUE_SIZE_IN_BYTES interesting?

It is not obvious how to compute it for CHAR. Probably some custom function will be needed. I decided to skip it for now. In the optimizer we can use the length from the type itself. Usually the deviation is not major (or otherwise there is less sense of using CHAR)

[findepi]

a TODO note?

[electrum]

Don't forget to add this TODO about *_VALUE_SIZE_IN_BYTES

[arhimondr]

Don't forget to add this TODO about *_VALUE_SIZE_IN_BYTES

I'm going to introduce it in a very next PR

[findepi]

why false? (generally MAX seems suitable and is used with returnFirstNonEmpty=true)

[arhimondr]

null can indicate 2 thins.

• statistic is missing (was never computed)
• In case of min/max - that the table is empty

If the table is empty NDV is not gonna be null, but 0.

[findepi]

if second is not present, you return empty
if second is present but has all fields absent, you return first

Why? Leave some explanation in the code.

(here & a few times below)

[arhimondr]

If only one of these is present - possibly the schema migration took place. If the column as a VARCHAR but become an INTEGER for example. Further we may want to do a proper schema migration here. I haven't decided how would it look like though.

[findepi]

So would something like this be proper?

// normally, either both or none is present

[findepi]

i am lost. we have ColumnStatisticType.NUMBER_OF_TRUE_VALUES but we don't have ColumnStatisticType.NUMBER_OF_FALSE_VALUES. Please explain

[arhimondr]

NUMBER_OF_FALSE_VALUES = NUMBER_OF_NON_NULL - NUMBER_OF_TRUE_VALUES

What we ask engine to compute does not necessary match the statistics we want to store.

[findepi]

if first is present, but second is not, we return first
if first is present, but second lacks row count, we return empty

Why? Leave some explanation in the code.

[arhimondr]

Changed to if (!firstRowCount.isPresent() || !secondRowCount.isPresent()) {. Hopefully it makes more sense.

[findepi]

OptionalDouble.of(0) ? In case we have ended up inserting zero rows into existing partition

[arhimondr]

If there 0 rows the average length is unknow. There is no rows. If we inserted values into empty partition it will go the first.isPresent() ? first : second; path.

[findepi]

I think I know why ((Number) type.getObjectValue(session, block, 0)).longValue() instead of type.getLong(block, 0) (because the second would work also for eg short decimal, skipping internal-representation-to-external conversion). While this method is called only for integral types, it's better as-is.
Consider leaving a comment hinting at the choice made

[findepi]

"Add properties for column statistics collect"

[findepi]

remove, boolean cannot be null

[findepi]

i wouldn't keep the empty line before this one

[findepi]

"Collect column statistics on table write: Documentation"

[findepi]

nit: move after REAL

[findepi]

MIN,MAX (if you support them in the code)

[arhimondr]

No, we do not collect MIN and MAX for Varchar

[findepi]

i am worried we may miss to update this line when changing the default of this property.
What about different wording, leveraging the property's default is in the table above

Automatic column level statistics collection on write is controlled by ``hive.collect-column-statistics-on-write`` property.

[findepi]

nit: why not just "MIN", "MAX"

[arhimondr]

So i can static-import it. MIN and MAX are used by the values from the ColumnStatisticType

[arhimondr]

Going to rename it back to MIN and MAX after i rename values in ColumnStatisticType

[arhimondr]

@rschlussel2 @findepi @kokosing Comments addressed

[arhimondr]

@findepi @kokosing @rschlussel @electrum

Heads up. I'm going to remove support for the MAX_VALUE_SIZE_IN_BYTES and AVERAGE_VALUE_SIZE_IN_BYTES statistics collection.

The reasons are next:

• Collecting of these statistics is inefficient. It requires additional projection.
• MAX_VALUE_SIZE_IN_BYTES is not used by the optimizer
• AVERAGE_VALUE_SIZE_IN_BYTES - our version of the Metastore stores this statistics as the IN_MEMORY_DATA_SIZE. Computing AVERAGE doesn't make much sense, as it is easier to compute total size, and than divide it by the number of rows.
• It is not possible to compute AVERAGE_VALUE_SIZE_IN_BYTES for the CHAR column with the existing aggregation functions.

I'm going to create a separate PR that:

• Introduces in_memory_data_size aggregation function for VARCHAR, VARBINARY, CHAR and complex types (MAP, ARRAY, ROW)
• Uses this function to compute IN_MEMORY_DATA_SIZE statistic
• Removes maxColumnLength from the HiveColumnStatistics
• Replaces averageColumnLength with inMemoryDataSizeInBytes
• Does the inMemoryDataSizeInBytes to averageColumnLength conversion in the ThriftMetastoreUtil

[findepi]

@arhimondr how does this play with #11107?
i understand, after all these PRs (current, #11107 and the one you're planning), we need to ensure interoperability:

• we should be able to read statistics computed by others (eg by Hive)
• we should be able to compute stats for ourselves
• we should be able to compute stats readable by other tools as well
• all this for a bunch of Hive Metastore versions

[electrum]

Almost all usages of throwIfInstanceOf are alone. The few places that it is combined with throwIfUnchecked should be converted to propagateIfPossible since that is more succinct.

[electrum]

propagateIfPossible(failure, Exception.class);

[electrum]

Nit: end Javadoc sentences in a period

[electrum]

Let's make this say "during a write" for now. We can update to "during a write or analyze" later when that is implemented.

[electrum]

Are there any concerns about memory size for Block? We have to store all of these in memory at once in the coordinator. (I'm not suggesting this is a problem, but rather asking if you've considered it or done any back-of-the-napkin calculations)

[electrum]

Nit: put min/max at the start or end, so that the NUMBER_* stats are together

[electrum]

I'm wondering if MAX should be MAX_VALUE. That might be more consistent with MAX_VALUE_SIZE_IN_BYTES.

[arhimondr]

I'm wondering if MAX should be MAX_VALUE. That might be more consistent with MAX_VALUE_SIZE_IN_BYTES.

Yeah. That it will be less chance that it clashes on static import. Let me change that.

[electrum]

Maybe wrap this to split up the different kinds of types

return type.equals(BIGINT) || type.equals(INTEGER) || type.equals(SMALLINT) || type.equals(TINYINT) ||
        type.equals(DOUBLE) || type.equals(REAL) ||
        type instanceof DecimalType;

[electrum]

We don't need to widen the table now that the default is "false"

[arhimondr]

Right. I'm staill going to widen it for 2 symbols to cover the RCBINARY

[electrum]

Should we use words here? This is for humans, and AFAIK, these constants like NUMBER_OF_NULLS are an internal detail of Presto.

Also, let's put min/max last, rather than in the middle of the "number of" stats.

``TINYINT``   number of nulls, number of distinct values, min/max values
``BOOLEAN``   number of nulls, number of true/false values

[electrum]

Automatic column level statistics collection on write is controlled by
the ``collect-column-statistics-on-write`` catalog session property.

[electrum]

Let's be consistent with other tables and put the "Collectible Statistics" at the start of the cell.

============= ====================================
Column Type   Collectible Statistics

[findepi]

Travis failure looks related:

2018-07-27 02:42:50 INFO: FAILURE     /    com.facebook.presto.tests.hive.TestTablePartitioningInsertInto.selectFromPartitionedNation (Groups: hive_connector, smoke) took 11.0 seconds
2018-07-27 02:42:50 SEVERE: Failure cause:
java.lang.AssertionError: 
Expecting:
 <0L>
to be equal to:
 <10L>
but was not.
	at com.facebook.presto.tests.hive.TestTablePartitioningInsertInto.testQuerySplitsNumber(TestTablePartitioningInsertInto.java:84)
	at com.facebook.presto.tests.hive.TestTablePartitioningInsertInto.selectFromPartitionedNation(TestTablePartitioningInsertInto.java:66)

[arhimondr]

Travis failure looks related:

It seems to be intermittent. I see no reason why this patch should've affected this test in any way.

[findepi]

hm...did you observe it failing intermittently on master as well?

[arhimondr]

hm...did you observe it failing intermittently on master as well?

At least once. I didn't look for more.

https://api.travis-ci.org/v3/job/400752461/log.txt (https://travis-ci.org/prestodb/presto/jobs/400752461, https://travis-ci.org/prestodb/presto/builds/400752451)

[wenleix]

Is there any reason not using ImmutableList.copyOf() ?

[arhimondr]

ImmutableList is a Guava class. We don't have Guava in the classpath of the presto-spi module.