Add support for push down filters in parquet sources

[isnotinvain]

This includes bumping parquet to version 1.6.0rc2

[ianoc]

have you tested this? even manually?

[johnynek]

Not crazy about the extra parameter group here. I see why you are doing it, but I am trying to think if we can find another way.

[jcoveney]

@johnynek: as far as I can tell there are 3 options. The parameter, a mutable internal variable, or f-bounded polymorphism.

[julienledem]

As there is only one class with this we could also do:
public constructor (fields: Fields, paths: String*)
private constructor (fields: Fields, paths: Array[String], filterPredicate: Option[FilterPredicate])
and add a
def withFilter(filter: FilterPredicate) = new FixedPathParquetTuple(fields, paths, filter)
From the user the syntax looks like:
new FixedPathParquetTuple(fields, "a", "b").withFilter(myFilter)

[jcoveney]

This is a great candidate for a platform test using the HadoopPlatformTest, though I haven't used parquet with that yet.

[julienledem]

@isnotinvain Could you add a test? It does not need to read a file, just instantiate the classes and check that the field is set. That would have the benefit of exercising the API

[isnotinvain]

@ianoc , @julienledem I'll add a unit test that makes sure the field is set correctly.

@jcoveney I think this functionality should be tested in parquet right? scalding is just passing things along?

As for the API, we went around on this a few times in the past, seems like we all want an immutable .withFilter method, which isn't really possible (even w/ f bounded polymorphism) without requiring each subclass to implement .withFilter

But there's only 3 subclasses, and likely to be less than 10 pretty much forever -- so how about I add the withFilter method and have each of these classes implement it (immutably via a copy constructor)

[julienledem]

I think you can omit (mf) here

[isnotinvain]

done!

[julienledem]

I restarted the stuck build and now it passes.

[tsdeng]

A little worry about calling the constructor here. User may extends this class, and calling copyWithFilter may return an instance of different class. And in the subclass, user may mix-in other trait or override some methods, and those changes will be missed after calling copyWithFilter.

[isnotinvain]

Actually, users should not override this class, maybe we should make it final.

Because if they override this class, the This parameter would be wrong.

[isnotinvain]

OK I updated to make the sources final, and instead of returning new anonymous classes, I pushed the values into fields in the class. This means you can do either:

val mySource = DailySuffixParquetThrift("/some/path", myDateRange, filterPredicate = Some(fp))

or you can do:

val mySource = DailySuffixParquetThrift("/some/path", myDateRange).withFilter(fp)

So, I didn't bother making the constructor that takes a filter predicate as an argument private. Seemed like overkill to make it private + provide an aux constructor, when that's really the same as having a default argument.

So at this point, the sources are final, immutable, and all support the .withFilter(fp) method. I think this is pretty good / ready to merge unless you have any more concerns.

Thanks!
Alex

[johnynek]

let's use a case match here:

val newFp = filterPredicate match {
  case None => fp
  case Some(old) => and(old, fp)
}

I feel like that is a clearer expression, and clearly perf does not matter here.

[isnotinvain]

done

[johnynek]

I think this looks great. Just some minor comments.

[johnynek]

merge when green.