Skip to content

[SPARK-26841][SQL] Kafka timestamp pushdown #23749

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
tomasbartalos wants to merge 6 commits into from
Closed

[SPARK-26841][SQL] Kafka timestamp pushdown #23749

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
21e2701
SPARK-26841 Timestamp pushdown to kafka relation with unit tests
tomasbartalos Feb 8, 2019
7af0c53
Merge branch 'master' into kafka-timestamp-pushdown
tomasbartalos Mar 15, 2019
4c68974
SPARK-26841 fixed compilation error on bad merge
tomasbartalos Mar 18, 2019
758097a
SPARK-26841 changes after Code Review
tomasbartalos Mar 20, 2019
ce01dae
Merge remote-tracking branch 'upstream/master' into kafka-timestamp-p…
tomasbartalos Apr 3, 2019
721c68e
SPARK-26841 resurrected mistakenly deleted test
tomasbartalos Apr 4, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -122,6 +122,16 @@ private[kafka010] class KafkaOffsetReader(
partitions.asScala.toSet
}

def fetchOffsetsByTime(times: Map[TopicPartition, Long]):
Map[TopicPartition, Option[Long]] = runUninterruptibly {
assert(Thread.currentThread().isInstanceOf[UninterruptibleThread])

consumer.offsetsForTimes(times.map{case (k, v) => k -> long2Long(v)}.asJava)
.asScala.map{case (k, v) =>
k -> (if (v != null) Some(Long2long(v.offset())) else None)
}.toMap
}

/**
* Resolves the specific offsets based on Kafka seek positions.
* This method resolves offset value -1 to the latest and -2 to the
Expand Down Expand Up @@ -394,6 +404,8 @@ private[kafka010] class KafkaOffsetReader(
}

private[kafka010] object KafkaOffsetReader {
// offsets are not instances of Optional, we need special state for None
val EMPTY_OFFSET: Long = -100L
Copy link
Contributor

@jose-torres jose-torres Mar 15, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure I follow why this is needed. Normally, if fetchOffsetsByTime sometimes needs to return offsets and sometimes needs to return None, it should just return Option[Offset] values.

Copy link
Author

@tomasbartalos tomasbartalos Mar 18, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Completely agree with you (this is what the comment is trying to express). I tried to go with least invasive changes, but since you've pointed this out, I can change the offset map everywhere from:
Map[TopicPartition, Long] to Map[TopicPartition, Option[Long]]

Copy link
Contributor

@jose-torres jose-torres Mar 18, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think it needs to be an Option[Long] everywhere. Just until we decide what value we want to pass to Kafka.

Copy link
Author

@tomasbartalos tomasbartalos Mar 20, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed to Option[Long] where it was necessary


def kafkaSchema: StructType = StructType(Seq(
StructField("key", BinaryType),
Expand Down
Loading