Add sort dataframe logic on qid

[atomic]

If qid is provided, xgb.DMatrix requires data to be sorted by qid - see data.cc:
https://github.com/ray-project/xgboost_ray/blob/master/xgboost_ray/matrix.py#L351

draft to add auto sorting of dataframe if qid is given and dataframe is not already sorted by qid

[Yard1]

I wonder if it wouldn't work better to include this logic in _get_dmatrix function in main.py. X and y there are pandas objects, so we can sort X and then apply the index to y. That way we support all cases as the sorting would happen on each actor right before a DMatrix is created.

[atomic]

@Yard1 I don't think we can sort cleanly in _get_dmatrix since the columns are already selected out of the dataframe to list.

[atomic]

Moved the logic in _split_dataframe in order to apply it to _DistributedRayDMatrixLoader as well.

[Yard1]

_split_dataframe works for me! Can we encapsulate the logic in a separate method for better modularity and call it in _split_dataframe? Thanks, let's see if CI is happy :)

Can we also add a test that would fail without sorted qids? The existing one can be modified to add a case like that.

[heyitsmui]

this is awesome! so this should be done at each worker, after each parquet is loaded into pdarray and concatenated into local_data and then you sort it by qid? After that you will use it to create dmatrix and do a ray.put to object store?

[heyitsmui]

@atomic can you also add a few unit test cases that cover our use case e.g. qids are across multiple parquet files and as long as we make sure per-worker level sorting works this will work?

[atomic]

yes, this sorting is done after ray concat all the actor shard's loaded dataframe into one. And yes, the sorted data is then put to ray object store.

[atomic]

@heyitsmui added integ tests for multi-parquet files

[Yard1]

can we raise an exception if it has more than 1 column?

[atomic]

sure

[Yard1]

Hey @atomic this is great! 2 small nits regarding tests. Can you also run format.sh script in the root repo folder to make lint CI pass? Thanks!

[Yard1]

Suggested change

	_ = DMatrix(**{
	"data": in_x,
	"label": in_y,
	"qid": unsorted_qid
	})
	_ = DMatrix(**{
	"data": in_x,
	"label": in_y,
	"qid": np.sort(unsorted_qid)
	}) # no exception
	# test RayDMatrix handles sorting automatically
	mat = RayDMatrix(in_x, in_y, qid=unsorted_qid)
	params = mat.get_data(rank=0, num_actors=1)
	_ = DMatrix(**params)
	DMatrix(**{
	"data": in_x,
	"label": in_y,
	"qid": unsorted_qid
	})
	DMatrix(**{
	"data": in_x,
	"label": in_y,
	"qid": np.sort(unsorted_qid)
	}) # no exception
	# test RayDMatrix handles sorting automatically
	mat = RayDMatrix(in_x, in_y, qid=unsorted_qid)
	params = mat.get_data(rank=0, num_actors=1)
	DMatrix(**params)

[Yard1]

Suggested change

	_ = DMatrix(**params)
	DMatrix(**params)

[Yard1]

@atomic xgboost_ray/matrix.py:61:80: E501 line too long (99 > 79 characters) (just this left!)

[Yard1]

Cutting edge test failure is the same as on master