refactor dataset stream