Build metadata endpoint and directory walker

[hanyuli1995]

Summary

We are supporting metadata upload to k-v store for key-value pair key->conf right now. We want to add a general class metadata endpoint to support more potential use cases.

This PR is to add two general class MetadataEndPoint and MetadataDirWalker

MetadataEndPoint:

case class MetadataEndPoint[Conf <: TBase[_, _]: Manifest: ClassTag](
    extractFn: (String, Conf) => (String, String),
    name: String
)

Defined with a extract function and an end point name. Extract function extracts the key-value pair from Conf(could be Join/GroupBy/StagingQuery) and file path(string). The name is the dataset name when we send the data to k-v store.

MetadataDirWalker:

class MetadataDirWalker(dirPath: String, metadataEndPointNames: List[String])

Go through the directory to iterate over all the config files and generate k-v pair metadata based on the metadata end points provided.

The PR adds two metadata endpoint ZIPLINE_METADATA and ZIPLINE_METADATA_BY_TEAM

CHRONON_METADATA:
key -> conf json in string format e.g : joins/team/team.example_join.v1 -> {...}

CHRONON_METADATA_BY_TEAM:
type/team -> list of key in string format e.g : joins/team -> a, b, c

Why / Goal

Test Plan

• Added Unit Tests
• Covered by existing CI
• Integration tested

Checklist

• Documentation update

Reviewers

[hanyuli1995]

We probably can make this a passed-in config, WDYT? @nikhilsimha

[nikhilsimha]

agreed - we could take this from conf

[hanyuli1995]

I am using the end point name as dataset here.

[nikhilsimha]

So we should make it so that we only parse the file once and run a bunch of extractors for the conf type.

So we need to maintain a list of (key, value) extractors for groupBys, joins and stagingQueries.

We simply parse once, run all the extractors per file, print it to console and write to metastore.

[nikhilsimha]

Added a high level comment

[better365]

Suggested change

	* ZIPLINE_METADATA_BY_TEAM -> (team -> List("join1", "join2")),
	* ZIPLINE_METADATA -> (teams/joins/join1 -> config1)
	* CHRONON_METADATA_BY_TEAM -> (team -> List("join1", "join2")),
	* CHRONON_METADATA -> (teams/joins/join1 -> config1)

[better365]

Suggested change

	val ConfByKeyEndPointName = "ZIPLINE_METADATA"
	val NameByTeamEndPointName = "ZIPLINE_METADATA_BY_TEAM"
	val ConfByKeyEndPointName = "CHRONON_METADATA"
	val NameByTeamEndPointName = "CHRONON_METADATA_BY_TEAM"

[better365]

Let's not use the old term zipline as much as we could.
Could you make those configurable or read from a config file or read from command line params?

Suggested change

	val acceptedEndPoints = List("ZIPLINE_METADATA")
	val acceptedEndPoints = List("CHRONON_METADATA")

[better365]

we could make this configurable.
We had some put request batch oversize error from our KV store before. That's why we set it to 50 for the safer side.

[hanyuli1995]

Yep it is a configurable parameter in put function. https://github.com/airbnb/chronon/blob/19664365ace064476d49b6830c980612fa2627d5/online/src/main/scala/ai/chronon/online/MetadataStore.scala#L154c I increase this to 100 because 50 doesn't work well for groupBy(there are about 1k groupBy and 200 joins. )

[better365]

why do we want this? If it is already bytes, what is the use of \n?

[hanyuli1995]

@better365 after the change the value could be a list of string. not sure what will be the value in k-v store if apply getBytes on a list instead of a string. I am trying to convert the list to a string first then apply toBytes

[better365]

I see. What will be used for the list of the configs? it will be easier to code it out here.

[hanyuli1995]

@better365 I update the dataset name from zipline to chronon now. For airbnb use case we also need to change the code in mussel otherwise the job will still fail. Will raise another PR in treehouse once this PR get stamped.

[better365]

Stamped. Please fix the failed UT.

[nikhilsimha]

this will break if there are commas in a single member string of the seq.

Chatgpt gave me this - I haven't test it myself.

import java.util.Base64
import java.nio.charset.StandardCharsets

object StringArrayConverter {
  // Method to convert an array of strings to a byte array using Base64 encoding for each element
  def stringsToBytes(strings: Array[String]): Array[Byte] = {
    val base64EncodedStrings = strings.map(s => Base64.getEncoder.encodeToString(s.getBytes(StandardCharsets.UTF_8)))
    base64EncodedStrings.mkString(",").getBytes(StandardCharsets.UTF_8)
  }

  // Method to convert a byte array back to an array of strings by decoding Base64
  def bytesToStrings(bytes: Array[Byte]): Array[String] = {
    val encodedString = new String(bytes, StandardCharsets.UTF_8)
    encodedString.split(",").map(s => new String(Base64.getDecoder.decode(s), StandardCharsets.UTF_8))
  }
}

// Testing the methods
val originalStrings = Array("a", "b,c,d", "e.f.g", "h|i|j")
val bytes = StringArrayConverter.stringsToBytes(originalStrings)
val decodedStrings = StringArrayConverter.bytesToStrings(bytes)

println("Original Strings: " + originalStrings.mkString("[", ", ", "]"))
println("Decoded Strings: " + decodedStrings.mkString("[", ", ", "]"))

[nikhilsimha]

I think we should break it out into its own test. There is a metadataexportertest.scala file

[hanyuli1995]

Hi @nikhilsimha I need to update the tests here because the original putConf method has been removed from metadataStore file.